Move cache globals to cache.py

Author: romainissynced

sync/clients/cache.py

@@ -1,6 +1,10 @@
 import logging
 from datetime import datetime, timedelta, timezone
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Union, Callable, Type
+from pathlib import Path
+import json
+
+from platformdirs import user_cache_dir
 
 logger = logging.getLogger(__name__)
 
@@ -43,3 +47,57 @@ def _set_cached_token(self) -> None:
 
     def _get_cached_token(self) -> Optional[Tuple[str, datetime]]:
         raise NotImplementedError
+
+
+class FileCachedToken(CachedToken):
+    def __init__(self):
+        self._cache_file = Path(user_cache_dir("syncsparkpy")) / "auth.json"
+
+        super().__init__()
+
+    def _get_cached_token(self) -> Optional[Tuple[str, datetime]]:
+        # Cache is optional, we can fail to read it and not worry
+        if self._cache_file.exists():
+            try:
+                cached_token = json.loads(self._cache_file.read_text())
+                cached_access_token = cached_token["access_token"]
+                cached_expiry = datetime.fromisoformat(cached_token["expires_at_utc"])
+                return cached_access_token, cached_expiry
+            except Exception as e:
+                logger.warning(
+                    f"Failed to read cached access token @ {self._cache_file}", exc_info=e
+                )
+
+        return None
+
+    def _set_cached_token(self) -> None:
+        # Cache is optional, we can fail to read it and not worry
+        try:
+            self._cache_file.parent.mkdir(parents=True, exist_ok=True)
+            self._cache_file.write_text(
+                json.dumps(
+                    {
+                        "access_token": self._access_token,
+                        "expires_at_utc": self._access_token_expires_at_utc.isoformat(),
+                    }
+                )
+            )
+        except Exception as e:
+            logger.warning(
+                f"Failed to write cached access token @ {self._cache_file}", exc_info=e
+            )
+
+
+# Putting this here instead of config.py because circular imports and typing.
+ACCESS_TOKEN_CACHE_CLS_TYPE = Union[Type[CachedToken], Callable[[], CachedToken]]
+_access_token_cache_cls: ACCESS_TOKEN_CACHE_CLS_TYPE = FileCachedToken  # Default to local file caching.
+
+
+def set_access_token_cache_cls(access_token_cache_cls: ACCESS_TOKEN_CACHE_CLS_TYPE) -> None:
+    global _access_token_cache_cls
+    _access_token_cache_cls = access_token_cache_cls
+
+
+def get_access_token_cache_cache() -> ACCESS_TOKEN_CACHE_CLS_TYPE:
+    global _access_token_cache_cls
+    return _access_token_cache_cls

sync/clients/sync.py

@@ -1,69 +1,15 @@
-import json
 import logging
-from datetime import datetime
-from pathlib import Path
-from typing import Generator, Optional, Tuple, Type, Callable, Union
+from typing import Generator, Optional
 
 import dateutil.parser
 import httpx
-from platformdirs import user_cache_dir
 
 from ..config import API_KEY, CONFIG, APIKey
 from . import USER_AGENT, RetryableHTTPClient, encode_json
-from .cache import CachedToken
-
+from .cache import ACCESS_TOKEN_CACHE_CLS_TYPE, FileCachedToken, get_access_token_cache_cache
 logger = logging.getLogger(__name__)
 
 
-class FileCachedToken(CachedToken):
-    def __init__(self):
-        self._cache_file = Path(user_cache_dir("syncsparkpy")) / "auth.json"
-
-        super().__init__()
-
-    def _get_cached_token(self) -> Optional[Tuple[str, datetime]]:
-        # Cache is optional, we can fail to read it and not worry
-        if self._cache_file.exists():
-            try:
-                cached_token = json.loads(self._cache_file.read_text())
-                cached_access_token = cached_token["access_token"]
-                cached_expiry = datetime.fromisoformat(cached_token["expires_at_utc"])
-                return cached_access_token, cached_expiry
-            except Exception as e:
-                logger.warning(
-                    f"Failed to read cached access token @ {self._cache_file}", exc_info=e
-                )
-
-        return None
-
-    def _set_cached_token(self) -> None:
-        # Cache is optional, we can fail to read it and not worry
-        try:
-            self._cache_file.parent.mkdir(parents=True, exist_ok=True)
-            self._cache_file.write_text(
-                json.dumps(
-                    {
-                        "access_token": self._access_token,
-                        "expires_at_utc": self._access_token_expires_at_utc.isoformat(),
-                    }
-                )
-            )
-        except Exception as e:
-            logger.warning(
-                f"Failed to write cached access token @ {self._cache_file}", exc_info=e
-            )
-
-
-# Putting this here instead of config.py because circular imports and typing.
-_access_token_cache_cls = FileCachedToken  # Default to local file caching.
-ACCESS_TOKEN_CACHE_CLS_TYPE = Union[Type[CachedToken], Callable[[], CachedToken]]
-
-
-def set_access_token_cache_cls(access_token_cache_cls: ACCESS_TOKEN_CACHE_CLS_TYPE) -> None:
-    global _access_token_cache_cls
-    _access_token_cache_cls = access_token_cache_cls
-
-
 class SyncAuth(httpx.Auth):
     requires_response_body = True
 
@@ -395,7 +341,7 @@ def get_default_client() -> SyncClient:
         _sync_client = SyncClient(
             CONFIG.api_url,
             API_KEY,
-            access_token_cache_cls=_access_token_cache_cls
+            access_token_cache_cls=get_access_token_cache_cache()
         )
     return _sync_client
 
@@ -409,6 +355,6 @@ def get_default_async_client() -> ASyncClient:
         _async_sync_client = ASyncClient(
             CONFIG.api_url,
             API_KEY,
-            access_token_cache_cls=_access_token_cache_cls
+            access_token_cache_cls=get_access_token_cache_cache()
         )
     return _async_sync_client