spotipy/spotipy/cache_handler.py
Tony Jackson 9550c8fd86
Create CacheHandler to abstract caching tokens (#625)
* Refactor functions into static methods of AuthBase

Functions is_token_expired was a loose function that was added to
SpotifyClientCredentials, SpotifyPKCE, and SpotifyImplicitGrant classes
through a method on each class that passed the call to the loose
is_token_expired function. Function _is_scope_subset was duplicated on
SpotifyClientCredentials, SpotifyPKCE, and SpotifyImplicitGrant classes.

Refactoring is_token_expired and _is_scope_subset to be static methods
on SpotifyAuthBase means both are available for all derived classes and
require less boilerplate.

* Create CacheHandler to abstract caching tokens

Previous code only supported caching to and from json files in a given
directory. In addition, the get_cached_token method mixed getting and
getting the token in the same method.

This change creates a CacheHandler class to abstract out the caching
implementation and allow the user to cache tokens in any way they
see fit. For example, the user could create a MongoCache class to store
and retrieve tokens from a Mongo database and specify that
cache_handler=MongoCache in creating an auth_manager object.

To implement the CacheHandler abstraction, the following changes are
implemented:

The validation code in each get_cached_token method in SpotifyOAuth,
SpotifyPKCE, and SpotifyImplicitGrant is moved into a validate_token
method in each class.

The CacheHandler class is created with get_cached_token and
save_token_to_cache methods.

Previous instances of self.get_cached_token() are now replaced with
self.validate_token(self.cache_handler.get_cached_token()) to preserve
the getting and validation behaviour.

cache_handler is added as an argument to SpotifyOAuth, SpotifyPKCE, and
SpotifyImplicitGrant. Specifying a cache_handler now overrides any
specification of cache_path and/or username.

To preserve backwards compatibility in handling cache files, a
CacheFileHandler class extending CacheHandler is created. If no
cache_handler is specified, the cache_path and username arguments are
used to create an instance of CacheFileHandler. It may be worth
deprecating the cache_path and username fields in favour of using
CacheFileHandler.

Tests are also modified and extended to cover the new functionality. A
sample MemoryCache CacheHandler is created to test getting and saving to
a custom CacheHandler.

* Fix cache_handler subclass check for Python 2

* Split assert message to fix line over max length

* Split cache handlers into cache_handler.py

* flake8 and autopep fixes

* Fix init to allow importing CacheHandler

When spotipy is installed as a package, CacheHandler is not accessible
from a `from spotipy import CacheHandler` statement because the import
is not specified in the __init__.py file. This commit adds CacheHandler
and CacheFileHandler to the init file so the user can import them.

* flake8 fix
2020-12-21 11:31:21 +01:00

85 lines
2.4 KiB
Python

__all__ = ['CacheHandler', 'CacheFileHandler']
import errno
import json
import logging
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()
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):
"""
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}`)
"""
if cache_path:
self.cache_path = cache_path
else:
cache_path = ".cache"
if username:
cache_path += "-" + str(username)
self.cache_path = cache_path
def get_cached_token(self):
token_info = None
try:
f = open(self.cache_path)
token_info_string = f.read()
f.close()
token_info = json.loads(token_info_string)
except IOError as error:
if error.errno == errno.ENOENT:
logger.debug("cache does not exist at: %s", self.cache_path)
else:
logger.warning("Couldn't read cache at: %s", self.cache_path)
return token_info
def save_token_to_cache(self, token_info):
try:
f = open(self.cache_path, "w")
f.write(json.dumps(token_info))
f.close()
except IOError:
logger.warning('Couldn\'t write token to cache at: %s',
self.cache_path)