__all__ = [ 'CacheHandler', 'CacheFileHandler', 'DjangoSessionCacheHandler', 'FlaskSessionCacheHandler', 'MemoryCacheHandler', 'RedisCacheHandler', 'MemcacheCacheHandler'] import errno import json import logging import os from redis import RedisError from spotipy.util import CLIENT_CREDS_ENV_VARS logger = logging.getLogger(__name__) class CacheHandler(): """ An abstraction layer for handling the caching and retrieval of authorization tokens. 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() def save_token_to_cache(self, token_info): """ Save a token_info dictionary object to the cache and return None. """ raise NotImplementedError() 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): """ 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 """ 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"])) if username: cache_path += "-" + str(username) self.cache_path = cache_path def get_cached_token(self): token_info = None try: with open(self.cache_path, encoding='utf-8') as f: token_info_string = f.read() token_info = json.loads(token_info_string) except OSError as error: if error.errno == errno.ENOENT: logger.debug(f"cache does not exist at: {self.cache_path}") else: logger.warning(f"Couldn't read cache at: {self.cache_path}") except json.JSONDecodeError: logger.warning(f"Couldn't decode JSON from cache at: {self.cache_path}") return token_info def save_token_to_cache(self, token_info): try: with open(self.cache_path, "w", encoding='utf-8') as f: f.write(json.dumps(token_info, cls=self.encoder_cls)) # https://github.com/spotipy-dev/spotipy/security/advisories/GHSA-pwhh-q4h6-w599 os.chmod(self.cache_path, 0o600) except OSError: logger.warning(f"Couldn't write token to cache at: {self.cache_path}") except FileNotFoundError: logger.warning(f"Couldn't set permissions to cache file at: {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. """ def __init__(self, token_info=None): """ Parameters: * token_info: The token info to store in memory. Can be None. """ self.token_info = token_info def get_cached_token(self): return self.token_info def save_token_to_cache(self, token_info): self.token_info = token_info class DjangoSessionCacheHandler(CacheHandler): """ A cache handler that stores the token info in the session framework provided by Django. Read more at https://docs.djangoproject.com/en/3.2/topics/http/sessions/ """ def __init__(self, request): """ Parameters: * request: HttpRequest object provided by Django for every incoming request """ self.request = request def get_cached_token(self): token_info = None try: token_info = self.request.session['token_info'] except KeyError: logger.debug("Token not found in the session") return token_info def save_token_to_cache(self, token_info): try: self.request.session['token_info'] = token_info except Exception as e: logger.warning(f"Error saving token to cache: {e}") class FlaskSessionCacheHandler(CacheHandler): """ A cache handler that stores the token info in the session framework provided by flask. """ def __init__(self, session): self.session = session def get_cached_token(self): token_info = None try: token_info = self.session["token_info"] except KeyError: logger.debug("Token not found in the session") return token_info def save_token_to_cache(self, token_info): try: self.session["token_info"] = token_info except Exception as e: logger.warning(f"Error saving token to cache: {e}") class RedisCacheHandler(CacheHandler): """ A cache handler that stores the token info in the Redis. """ def __init__(self, redis, key=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`) """ self.redis = redis self.key = key if key else 'token_info' def get_cached_token(self): token_info = None try: token_info = self.redis.get(self.key) if token_info: return json.loads(token_info) except RedisError as e: logger.warning(f"Error getting token from cache: {e}") return token_info def save_token_to_cache(self, token_info): try: self.redis.set(self.key, json.dumps(token_info)) except RedisError as e: logger.warning(f"Error saving token to cache: {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: * memcache: memcache client object provided by pymemcache (https://pymemcache.readthedocs.io/en/latest/getting_started.html) * key: May be supplied, will otherwise be generated (takes precedence over `token_info`) """ self.memcache = memcache self.key = key if key else 'token_info' def get_cached_token(self): from pymemcache import MemcacheError try: token_info = self.memcache.get(self.key) if token_info: return json.loads(token_info.decode()) except MemcacheError as e: logger.warning(f"Error getting token to cache: {e}") def save_token_to_cache(self, token_info): from pymemcache import MemcacheError try: self.memcache.set(self.key, json.dumps(token_info)) except MemcacheError as e: logger.warning(f"Error saving token to cache: {e}")