Add type annotations to cache_handler.py

CHANGELOG.md

@@ -18,6 +18,7 @@ Add your changes below.
 - Added custom `urllib3.Retry` class for printing a warning when a rate/request limit is reached.
 - Added `personalized_playlist.py`, `track_recommendations.py`, and `audio_features_analysis.py` to `/examples`.
 - Discord badge in README
+- Type annotations to `spotipy.cache_handler`
 
 ### Fixed
 - Audiobook integration tests

pyproject.toml

@@ -0,0 +1,35 @@
+[project]
+name = "spotipy"
+version = "2.24.0"
+description = "A light weight Python library for the Spotify Web API"
+authors = [
+    { name = "@plamere", email = "[email protected]" }
+]
+readme = "README.md"
+urls = { "homepage" = "https://spotipy.readthedocs.org/", "source" = "https://github.com/plamere/spotipy" }
+license = { text = "MIT" }
+requires-python = ">=3.8"
+dependencies = [
+    "redis>=3.5.3",
+    "requests>=2.25.0",
+    "urllib3>=1.26.0",
+    "pymemcache>=4.0.0",
+]
+
+
+[project.optional-dependencies]
+memcache = ["pymemcache>=3.5.2"]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.rye]
+dev-dependencies = [
+    "mypy>=1.11.1",
+    "pylint>=3.2.6",
+    "flake8>=7.1.1",
+    "isort>=5.13.2",
+    "autopep8>=2.3.1",
+]
+

requirements-dev.lock

@@ -0,0 +1,61 @@
+# generated by rye
+# use `rye lock` or `rye sync` to update this lockfile
+#
+# last locked with the following flags:
+#   pre: false
+#   features: []
+#   all-features: false
+#   with-sources: false
+#   generate-hashes: false
+#   universal: false
+
+-e file:.
+astroid==3.2.4
+    # via pylint
+async-timeout==4.0.3
+    # via redis
+autopep8==2.3.1
+certifi==2024.7.4
+    # via requests
+charset-normalizer==3.3.2
+    # via requests
+dill==0.3.8
+    # via pylint
+flake8==7.1.1
+idna==3.7
+    # via requests
+isort==5.13.2
+    # via pylint
+mccabe==0.7.0
+    # via flake8
+    # via pylint
+mypy==1.11.1
+mypy-extensions==1.0.0
+    # via mypy
+platformdirs==4.2.2
+    # via pylint
+pycodestyle==2.12.1
+    # via autopep8
+    # via flake8
+pyflakes==3.2.0
+    # via flake8
+pylint==3.2.6
+pymemcache==4.0.0
+    # via spotipy
+redis==5.0.8
+    # via spotipy
+requests==2.32.3
+    # via spotipy
+tomli==2.0.1
+    # via autopep8
+    # via mypy
+    # via pylint
+tomlkit==0.13.0
+    # via pylint
+typing-extensions==4.12.2
+    # via astroid
+    # via mypy
+    # via pylint
+urllib3==2.2.2
+    # via requests
+    # via spotipy

requirements.lock

@@ -0,0 +1,29 @@
+# generated by rye
+# use `rye lock` or `rye sync` to update this lockfile
+#
+# last locked with the following flags:
+#   pre: false
+#   features: []
+#   all-features: false
+#   with-sources: false
+#   generate-hashes: false
+#   universal: false
+
+-e file:.
+async-timeout==4.0.3
+    # via redis
+certifi==2024.7.4
+    # via requests
+charset-normalizer==3.3.2
+    # via requests
+idna==3.7
+    # via requests
+pymemcache==4.0.0
+    # via spotipy
+redis==5.0.8
+    # via spotipy
+requests==2.32.3
+    # via spotipy
+urllib3==2.2.2
+    # via requests
+    # via spotipy

spotipy/cache_handler.py

@@ -1,79 +1,81 @@
-__all__ = [
-    'CacheHandler',
-    'CacheFileHandler',
-    'DjangoSessionCacheHandler',
-    'FlaskSessionCacheHandler',
-    'MemoryCacheHandler',
-    'RedisCacheHandler',
-    'MemcacheCacheHandler']
+from __future__ import annotations
 
 import errno
 import json
 import logging
 import os
-from spotipy.util import CLIENT_CREDS_ENV_VARS
-
+from abc import ABC, abstractmethod
+from json import JSONEncoder
+from typing import TypedDict
+import redis
 from redis import RedisError
+import redis.client
+
+from .util import CLIENT_CREDS_ENV_VARS
+
+__all__ = [
+    "CacheHandler",
+    "CacheFileHandler",
+    "DjangoSessionCacheHandler",
+    "FlaskSessionCacheHandler",
+    "MemoryCacheHandler",
+    "RedisCacheHandler",
+    "MemcacheCacheHandler",
+]
 
 logger = logging.getLogger(__name__)
 
 
-class CacheHandler():
-    """
-    An abstraction layer for handling the caching and retrieval of
-    authorization tokens.
+class TokenInfo(TypedDict):
+    access_token: str
+    token_type: str
+    expires_in: int
+    scope: str
+    expires_at: int
+    refresh_token: str
 
-    Custom extensions of this class must implement get_cached_token
-    and save_token_to_cache methods with the same input and output
-    structure as the CacheHandler class.
-    """
 
-    def get_cached_token(self):
-        """
-        Get and return a token_info dictionary object.
-        """
-        # return token_info
-        raise NotImplementedError()
+class CacheHandler(ABC):
+    """An ABC for handling the caching and retrieval of authorization tokens."""
 
-    def save_token_to_cache(self, token_info):
-        """
-        Save a token_info dictionary object to the cache and return None.
-        """
-        raise NotImplementedError()
-        return None
+    @abstractmethod
+    def get_cached_token(self) -> TokenInfo | None:
+        """Get and return a token_info dictionary object."""
 
+    @abstractmethod
+    def save_token_to_cache(self, token_info: TokenInfo) -> None:
+        """Save a token_info dictionary object to the cache and return None."""
 
-class CacheFileHandler(CacheHandler):
-    """
-    Handles reading and writing cached Spotify authorization tokens
-    as json files on disk.
-    """
 
-    def __init__(self,
-                 cache_path=None,
-                 username=None,
-                 encoder_cls=None):
+class CacheFileHandler(CacheHandler):
+    """Read and write cached Spotify authorization tokens as json files on disk."""
+
+    def __init__(
+        self,
+        cache_path: str | None = None,
+        username: str | None = None,
+        encoder_cls: type[JSONEncoder] | None = None,
+    ) -> None:
         """
-        Parameters:
-             * cache_path: May be supplied, will otherwise be generated
-                           (takes precedence over `username`)
-             * username: May be supplied or set as environment variable
-                         (will set `cache_path` to `.cache-{username}`)
-             * encoder_cls: May be supplied as a means of overwriting the
-                        default serializer used for writing tokens to disk
+        Initialize CacheFileHandler instance.
+
+        :param cache_path: (Optional) Path to cache. (Will override 'username')
+        :param username: (Optional) Client username. (Can also be supplied via env var.)
+        :param encoder_cls: (Optional) JSON encoder class to override default.
         """
         self.encoder_cls = encoder_cls
         if cache_path:
             self.cache_path = cache_path
         else:
             cache_path = ".cache"
-            username = (username or os.getenv(CLIENT_CREDS_ENV_VARS["client_username"]))
+            username = username or os.getenv(CLIENT_CREDS_ENV_VARS["client_username"])
             if username:
                 cache_path += "-" + str(username)
             self.cache_path = cache_path
 
-    def get_cached_token(self):
-        token_info = None
+    def get_cached_token(self) -> TokenInfo | None:
+        """Get cached token from file."""
+        token_info: TokenInfo | None = None
 
         try:
             f = open(self.cache_path)
@@ -89,34 +91,33 @@ def get_cached_token(self):
 
         return token_info
 
-    def save_token_to_cache(self, token_info):
+    def save_token_to_cache(self, token_info: TokenInfo) -> None:
+        """Save token cache to file."""
         try:
             f = open(self.cache_path, "w")
             f.write(json.dumps(token_info, cls=self.encoder_cls))
             f.close()
         except OSError:
-            logger.warning('Couldn\'t write token to cache at: %s',
-                           self.cache_path)
+            logger.warning("Couldn't write token to cache at: %s", self.cache_path)
 
 
 class MemoryCacheHandler(CacheHandler):
-    """
-    A cache handler that simply stores the token info in memory as an
-    instance attribute of this class. The token info will be lost when this
-    instance is freed.
-    """
+    """Cache handler that stores the token non-persistently as an instance attribute."""
 
-    def __init__(self, token_info=None):
+    def __init__(self, token_info: TokenInfo | None = None) -> None:
         """
-        Parameters:
-            * token_info: The token info to store in memory. Can be None.
+        Initialize MemoryCacheHandler instance.
+
+        :param token_info: Optional initial cached token
         """
         self.token_info = token_info
 
-    def get_cached_token(self):
+    def get_cached_token(self) -> TokenInfo | None:
+        """Retrieve the cached token from the instance."""
         return self.token_info
 
-    def save_token_to_cache(self, token_info):
+    def save_token_to_cache(self, token_info: TokenInfo) -> None:
+        """Cache the token in this instance."""
         self.token_info = token_info
 
 
@@ -178,42 +179,42 @@ def save_token_to_cache(self, token_info):
 
 
 class RedisCacheHandler(CacheHandler):
-    """
-    A cache handler that stores the token info in the Redis.
-    """
+    """A cache handler that stores the token info in the Redis."""
 
-    def __init__(self, redis, key=None):
+    def __init__(self, redis_obj: redis.client.Redis, key: str | None = None) -> None:
         """
-        Parameters:
-            * redis: Redis object provided by redis-py library
-            (https://github.com/redis/redis-py)
-            * key: May be supplied, will otherwise be generated
-                   (takes precedence over `token_info`)
+        Initialize RedisCacheHandler instance.
+
+        :param redis: The Redis object to function as the cache
+        :param key: (Optional) The key to used to store the token in the cache
         """
-        self.redis = redis
-        self.key = key if key else 'token_info'
+        self.redis = redis_obj
+        self.key = key if key else "token_info"
 
-    def get_cached_token(self):
+    def get_cached_token(self) -> TokenInfo | None:
+        """Fetch cache token from the Redis."""
         token_info = None
         try:
             token_info = self.redis.get(self.key)
-            if token_info:
-                return json.loads(token_info)
+            if token_info is not None:
+                token_info = json.loads(token_info)
         except RedisError as e:
-            logger.warning('Error getting token from cache: ' + str(e))
+            logger.warning("Error getting token from cache: %s", str(e))
 
         return token_info
 
-    def save_token_to_cache(self, token_info):
+    def save_token_to_cache(self, token_info: TokenInfo) -> None:
+        """Cache token in the Redis."""
         try:
             self.redis.set(self.key, json.dumps(token_info))
         except RedisError as e:
-            logger.warning('Error saving token to cache: ' + str(e))
+            logger.warning("Error saving token to cache: %s", str(e))
 
 
 class MemcacheCacheHandler(CacheHandler):
     """A Cache handler that stores the token info in Memcache using the pymemcache client
     """
+
     def __init__(self, memcache, key=None) -> None:
         """
         Parameters:

spotipy/util.py

@@ -153,6 +153,7 @@ class Retry(urllib3.Retry):
     """
     Custom class for printing a warning when a rate/request limit is reached.
     """
+
     def increment(
             self,
             method: str | None = None,