Skip to content

Instantly share code, notes, and snippets.

@HacKanCuBa HacKanCuBa/cached.py
Last active Nov 8, 2019

Embed
What would you like to do?
Django Cached: simple cache abstract classes to create and use cached objects.
"""Handle object caching and data retrieval from API endpoints.
These abstract classes makes it easy to use Django's cache with a custom
object, and are very flexible. It support slots natively, and logs cache
hits/misses.
:Requirements:
- Django 2.0+
- Python 3.6+
:Example:
>>> class CustomObject(AbstractCachedBase):
>>>
>>> def __init__(self, **kwargs):
>>> self.a = None # type: Optional[int]
>>> self.b = '' # type: str
>>> self.from_dict(kwargs)
>>>
>>> @property
>>> def _cached_uid(self) -> str:
>>> return '' if self.a is None else str(self.a)
>>>
>>>
>>> class OtherCustomObject(AbstractCachedBase):
>>>
>>> def __init__(self):
>>> self.a = None # type: Optional[bool]
>>>
>>> @property
>>> def _cached_uid(self) -> str:
>>> return ''
>>>
>>>
>>> class FromAPI(AbstractCached):
>>>
>>> def __init__(self):
>>> self.a = None # type: Optional[int]
>>> ...
>>>
>>> def _get_data_from_endpoint(self) -> dict:
>>> # Just define how data is retrieved from your endpoint and make sure
>>> # to return it as dict. Then AbstractCached will do its magic :)
>>> data = {}
>>> response = requests.get(
>>> f'https://myendpoint.local/api/v1/item/{self.a}',
>>> headers={'content-type': 'application/json'},
>>> timeout=5,
>>> )
>>> if response.ok:
>>> data = response.json()
>>> return data
>>>
>>> @property
>>> def _cached_uid(self) -> str:
>>> return str(self.a) # Note that str(None) == 'None'
>>>
>>>
>>> custom_obj = CustomObject(a=2019) # Instantiate, populate
>>> custom_obj.b = '#FreeChelsea #FreeAssange'
>>> print(custom_obj.as_dict) # {'a': 2019, 'b': '#FreeChelsea #FreeAssange'}
>>> custom_obj.save() # Store in cache
>>>
>>> custom_obj2 = CustomObject(a=2019)
>>> custom_obj2.cache_retrieve() # Get from cache
>>> print(custom_obj2.as_dict) # {'a': 2019, 'b': '#FreeChelsea #FreeAssange'}
>>>
>>> custom_obj3 = CustomObject(b='#FreeOlaBini') # Instantiate with empty unique id
>>> custom_obj3.save()
>>> other_custom_obj = OtherCustomObject() # It also has empty unique identifier
>>> other_custom_obj.a = True
>>> other_custom_obj.save()
>>> # Let's see what happens, given that both identifiers are empty
>>> other_custom_obj2 = OtherCustomObject()
>>> other_custom_obj2.cache_retrieve()
>>> custom_obj4 = CustomObject()
>>> custom_obj4.cache_retrieve()
>>> print(other_custom_obj2.as_dict) # {'a': True}
>>> print(custom_obj4.as_dict) # {'a': None, 'b': '#FreeOlaBini'}
>>> # Different classes use different cache identifiers, so they never collide even
>>> # if their unique identifier is the same.
>>>
>>> myremoteobj = FromAPI()
>>> myremoteobj.a = 1
>>> myremoteobj.retrieve() # Fetch data from cache or, if it misses, from the endpoint
>>> myremoteobj.retrieve(invalidate_cache=True) # Fetch from the endpoint forcefully
"""
import logging
from abc import ABC
from abc import abstractmethod
from hashlib import blake2b # md5 could be used here w/o harm
from typing import Generator
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from django.core.cache import DEFAULT_CACHE_ALIAS
from django.core.cache import caches
__version__ = '0.7.1'
__author__ = 'HacKan (https://hackan.net)'
__license__ = 'GPL-3+'
__url__ = 'https://gist.github.com/HacKanCuBa/b14b79ca88b8a1e999be9c0ba1fd878c'
logger = logging.getLogger(__name__)
class CachedException(Exception):
"""Base exception for this lib."""
class CachedUnsupportedException(CachedException):
"""Exception for unsupported operation."""
class AbstractCachedBaseMinimal(ABC):
"""Abstract base class with a handful of helper methods."""
def __bool__(self) -> bool:
"""Get the boolean representation of the object."""
return any(self.as_dict.values())
@property
def _attributes(self) -> Tuple[str, ...]:
"""Get a tuple of all object's attributes alphabetically ordered.
Attributes are exported as is. It considers both slots and otherwise.
Do not override this method.
"""
# __slots__ and __dict__ might coexist, where a parent might define
# __slots__, and a child not (thus using __dict__)
attrs = set() # type: Set[str]
if hasattr(self, '__slots__'):
for cls in self.__class__.__mro__:
attrs.update(getattr(cls, '__slots__', ()))
if hasattr(self, '__dict__'):
attrs.update(self.__dict__.keys())
return tuple(sorted(attrs))
@property
def _attributes_public(self) -> Tuple[str, ...]:
"""Get a tuple of object's public attributes only, alphabetically ordered.
It considers both slots and otherwise. Private attributes are not
exported at all. This means that private attributes that have public
counterparts are exported solely as the public counterpart, whereas
private without public counterparts are not exported.
Do not override this method.
"""
attrs = self._attributes
pub_attrs = set(filter(lambda attr: attr[0] != '_', attrs))
pub_attrs.update(attr[1:] for attr in attrs
if attr[0] == '_' and hasattr(self.__class__, attr[1:]))
return tuple(sorted(pub_attrs))
def _get_attributes(self) -> Tuple[str, ...]:
"""Get the object's attributes alphabetically ordered, using getters if any.
This means that private attributes that have public counterparts are
exported solely as the public counterpart, whereas private without
public counterparts are exported as is.
"""
return tuple(sorted({
attr[1:] if attr[0] == '_' and hasattr(self.__class__, attr[1:])
else attr
for attr in self._attributes
}))
@property
def as_dict(self) -> dict:
"""Get the representation of the object as a dictionary."""
return {key: getattr(self, key) for key in self._attributes
if hasattr(self, key)}
def from_dict(self, data: dict, *, force: bool = False) -> None:
"""Set the object properties from a dictionary.
:param data: Dictionary to read.
:param force: Force setting values from the dict by creating attributes
if necessary (it won't work with slots!).
"""
if data: # Prevent data being None
own_keys = self._attributes
for key, value in data.items():
if force or key in own_keys:
setattr(self, key, value)
class AbstractCachedInterface(ABC):
"""Abstract interface with some common cache operations.
:Usage:
Extend and optionally define `CACHE_NAME`, `CACHE_VERSION`, and `CACHE_TTL`.
:Attributes:
CACHE_NAME str: Name of the cache to use (defaults to Django's
default cache alias).
CACHE_VERSION int: The default version number for cache keys (defaults
to 1).
CACHE_TTL int: The number of seconds before a cache entry is
considered stale. If the value is None, cache
entries will not expire. If the value is 0, cache
entries immediately expire (effectively “don’t
cache”) (defaults to Django's value 300).
"""
# Name of the cache to use
CACHE_NAME = DEFAULT_CACHE_ALIAS # type: str
# The default version number for cache keys.
CACHE_VERSION = 1 # type: int
# The number of seconds before a cache entry is considered stale. If the
# value is None, cache entries will not expire. If the value is 0, cache
# entries immediately expire (effectively “don’t cache”).
CACHE_TTL = 300 # type: Optional[int]
@property
def _caches(self):
"""Interface for Django `caches`."""
return caches
@property
def _cache(self):
"""Interface for the selected cache."""
return self._caches[self.CACHE_NAME]
@classmethod
def _cached_build_key(cls, key: str) -> str:
"""Build a cache key based on the given parameter.
Additionally uses the class name. If the key was already built, it
returns it.
It's heavily recommended to use this along `_cached_encode_key` to
obtain a safe cache key (some cache backends have troubles with key
lengths and non-ascii characters).
:param key: Key to build for the cache.
"""
key_prefix = f'Cached:{cls.__name__}:'
if key.startswith(key_prefix):
return key
return f'{key_prefix}{key}'
@staticmethod
def _cached_encode_key(key: str) -> str:
"""Encode a key as hexadecimal chars (a-f, 0-9) for safe usage.
The encoding is done by hashing the key (assuming UTF-8 string).
This is meant to be used by passing the encoded key to `_cached_build_key`
to obtain a safe cache key (some cache backends have troubles with key
lengths and non-ascii characters).
"""
return blake2b(key.encode('utf-8'), digest_size=16).hexdigest()
def _cached_interface_keys(self, pattern: str) -> List[str]:
"""Interface for `keys`. Execute KEYS command and return matched results.
The pattern used is generated by passing `pattern` to `_cached_build_key`.
:param pattern: The pattern passed to `_cached_build_key` to generate the
key pattern.
:raises CachedUnsupportedException: The cache backend doesn't support
searching for keys.
"""
key_pattern = self._cached_build_key(pattern)
try:
return self._cache.keys(key_pattern, version=self.CACHE_VERSION)
except AttributeError:
raise CachedUnsupportedException('The cache backend does not '
'support searching for keys')
def _cached_interface_iter_keys(self, pattern: str) -> Generator:
"""Interface for `iter_keys`. Same as keys, but uses redis >= 2.8 cursors.
The pattern used is generated by passing `pattern` to `_cached_build_key`.
:param pattern: The pattern passed to `_cached_build_key` to generate the
key pattern.
:raises CachedUnsupportedException: The cache backend doesn't support
searching for keys.
"""
key_pattern = self._cached_build_key(pattern)
try:
return self._cache.iter_keys(key_pattern, version=self.CACHE_VERSION)
except AttributeError:
raise CachedUnsupportedException('The cache backend does not '
'support searching for keys')
def _cached_interface_get(self, key: str) -> any:
"""Interface for `get`. Retrieve a value from the cache.
:param key: Key to fetch, which is composed by passing this value to
`_cached_build_key`.
"""
cache_key = self._cached_build_key(key)
return self._cache.get(cache_key, version=self.CACHE_VERSION)
def _cached_interface_set(self, key: str, value: any) -> None:
"""Interface for `set`. Persist a value to the cache.
:param key: Key to fetch, which is composed by passing this value to
`_cached_build_key`.
:param value: Value to store.
"""
cache_key = self._cached_build_key(key)
self._cache.set(cache_key, value, timeout=self.CACHE_TTL,
version=self.CACHE_VERSION)
def _cached_interface_delete(self, key: str) -> Optional[int]:
"""Interface for `delete`. Remove a key from the cache.
:param key: Key to delete, which is composed by passing this value to
`_cached_build_key`.
"""
cache_key = self._cached_build_key(key)
return self._cache.delete(cache_key, version=self.CACHE_VERSION)
def _cached_interface_delete_many(self, keys: List[str]) -> Optional[int]:
"""Interface for `delete_many`. Remove multiple keys at once.
:param keys: Keys to delete, which are composed by passing each value to
`_cached_build_key`.
"""
cache_keys = {self._cached_build_key(key) for key in keys}
return self._cache.delete_many(cache_keys, version=self.CACHE_VERSION)
def _cached_interface_delete_pattern(self, pattern: str) -> Optional[int]:
"""Interface for `delete_pattern`. Remove all keys matching pattern.
:param pattern: The pattern passed to `_cached_build_key` to generate the
key pattern.
:raises CachedUnsupportedException: The cache backend doesn't support
searching for keys.
"""
key_pattern = self._cached_build_key(pattern)
try:
return self._cache.delete_pattern(key_pattern, version=self.CACHE_VERSION)
except AttributeError:
raise CachedUnsupportedException('The cache backend does not '
'support searching for keys')
class AbstractCachedBaseGeneric(AbstractCachedBaseMinimal, AbstractCachedInterface):
"""Abstract base class with generic methods for handling cache.
:Usage:
Extend and define the following methods:
_cached_uid() -> str:
Return a unique identifier for the object in the cache as string.
May contain any character and any number of characters. Its hash is
used as an actual key.
"""
@property
@abstractmethod
def _cached_uid(self) -> str:
"""Define the unique object identifier.
This value must uniquely represent `self` amongst others of the same
class.
"""
pass
@property
def _cached_key_uid(self) -> str:
"""Get the unique object identifier encoded to be used as key.
It is `_cached_uid` as a 16 characters fixed length encoded hexadecimal
lowercase string.
"""
# To avoid character issues with cache key, hash it.
return self._cached_encode_key(self._cached_uid)
@property
def _cached_key(self) -> str:
"""Get the built cache key for the object.
Implements `_cached_build_key` with `_cached_key_uid`, the unique
encoded parameter for the object.
"""
return self._cached_build_key(self._cached_key_uid)
def __str__(self) -> str:
"""Get the readable string representation of the object."""
return self._cached_uid
def __repr__(self) -> str:
"""Get the unambiguous string representation of the object."""
return (
f'{self.__class__.__name__}('
f'cache={self.CACHE_NAME}, '
f'version={self.CACHE_VERSION}, '
f'key={self._cached_key}, '
f'stored={self._cached_key in self._cache})'
)
def _cache_retrieve_any_one(self, key_pattern: str = '*') -> bool:
"""Retrieve any one object of the same class from the cache.
This method searches for keys of the same class as `self` for the given
key pattern and retrieves the first one found. If no key pattern is
specified, but rather a full key, then retrieves that specific key
without searching.
:param key_pattern: Pattern to search (it can also be an exact value).
:return: True for cache hit, False otherwise.
:raises CachedUnsupportedException: The cache backend doesn't support
searching for keys.
"""
key = self._cached_build_key(key_pattern)
if '*' in key_pattern:
keys_iter = self._cached_interface_iter_keys(key_pattern)
else:
keys_iter = iter([key]) if key in self._cache else iter([])
try:
key = next(keys_iter)
except StopIteration:
hit = False
else:
hit = True
self.from_dict(self._cached_interface_get(key))
logger.debug(
'Cache %s in %s cache (v%d) for key %s',
'HIT' if hit else 'MISS',
self.CACHE_NAME,
self.CACHE_VERSION,
key,
)
return hit
def _cache_retrieve(self) -> bool:
"""Retrieve `self` from the cache.
:return: True for cache hit, False for miss.
"""
return self._cache_retrieve_any_one(self._cached_key)
def _cache_save(self) -> None:
"""Store the object in the cache."""
self._cached_interface_set(self._cached_key, self.as_dict)
logger.debug(
'Cache SET in %s cache (v%d) for key %s',
self.CACHE_NAME,
self.CACHE_VERSION,
self._cached_key,
)
def _cache_delete(self, key_pattern: str, *,
raise_exception: bool = False) -> Optional[int]:
"""Delete object(s) of the same class as `self` from the cache.
May return the number of deleted objects or None depending on the cache
backend.
Note: some backends doesn't support searching for keys. In such cases,
an exception could be raised if raise_exception is True, otherwise is
silently passed.
:param key_pattern: Key pattern to delete from the cache.
If it contains an asterisk (*), a search is
performed for several keys and all of them are
deleted (if the backend supports it).
:param raise_exception: True to raise an exception when the cache
backend doesn't support searching for keys.
:return: The number of objects deleted from the cache (0 or more) or None.
:raises CachedUnsupportedException: The cache backend doesn't support
searching for keys.
"""
if '*' in key_pattern:
try:
count = self._cached_interface_delete_pattern(key_pattern)
except CachedUnsupportedException:
if raise_exception:
raise
logger.exception('Cache %s does not support searching for keys',
self.CACHE_NAME)
count = 0
else:
# To be able to delete `self` from cache in backends that doesn't
# implement key searching
count = self._cached_interface_delete(key_pattern)
logger.debug(
'Cache DELETE in %s cache (v%d) for key %s: %s',
self.CACHE_NAME,
self.CACHE_VERSION,
self._cached_build_key(key_pattern),
f'{count}',
)
return count
class AbstractCachedBase(AbstractCachedBaseGeneric):
"""Abstract base class to handle cache storage of data.
:Usage:
Extend and define the following methods:
_cached_uid() -> str:
Return a unique identifier for the object in the cache as string.
May contain any character and any number of characters. Its hash is
used as an actual key.
Preferably also set the attributes CACHE_TTL and CACHE_VERSION. If you
need to use a different cache from the default, set CACHE_NAME.
:Object usage:
Instantiate an object, set its properties and call cache_retrieve() to
fetch object data from cache. Call save() to store object properties.
"""
@property
@abstractmethod
def _cached_uid(self) -> str:
"""Define the unique object identifier.
This value must uniquely represent `self` amongst others of the same
class.
"""
pass
def save(self) -> None:
"""Store itself in the cache."""
self._cache_save()
def cache_retrieve(self) -> bool:
"""Retrieve itself from the cache.
:return True for cache hit, False otherwise.
"""
return self._cache_retrieve()
def cache_delete(self) -> Optional[int]:
"""Delete itself from the cache.
May return the number of deleted objects or None depending on the cache
backend.
"""
return self._cache_delete(self._cached_key)
def cache_clear(self, *, raise_exception: bool = True) -> Optional[int]:
"""Delete every object of the same class as itself from the cache.
May return the number of deleted objects (0 or more) or None depending
on the cache backend.
Note: some backends doesn't support searching for keys. In such cases,
an exception is raised unless raise_exception is False.
:param raise_exception: True to raise an exception when the cache
backend doesn't support searching for keys.
:return: The number of objects deleted from the cache (0 or more) or None.
:raises CachedUnsupportedException: The cache backend doesn't support
searching for keys.
"""
return self._cache_delete('*', raise_exception=raise_exception)
class AbstractCached(AbstractCachedBase):
"""Abstract class to handle cache storage and retrieval of data.
:Usage:
Extend and define the following methods:
_cached_uid() -> str:
Return a unique identifier for the object in the cache as string.
May contain any character and any number of characters. Its hash is
used as an actual key.
_get_data_from_endpoint() -> dict:
Connect to the endpoint and retrieve desired data, returning it as
a dict to be cached and used with from_dict() method.
Preferably also set the attributes CACHE_TTL and CACHE_VERSION. If you
need to use a different cache from the default, set CACHE_NAME.
:Object usage:
Instantiate an object, set its properties and call retrieve() to fetch
object data from an endpoint (setting cache) or cache. If you don't
want to retrieve and instead set the properties, call save() to store
it in the cache.
"""
@abstractmethod
def _get_data_from_endpoint(self) -> dict:
"""Retrieve object data from the endpoint."""
pass
def unserialize(self, data: dict) -> None:
"""Read data from a dictionary and set the object properties.
Will try to set attributes from the dictionary, silently skipping those
that couldn't be set.
To transform data, creating a setter for the attribute is the recommended
way.
:param data: Dictionary to read.
"""
if data: # Prevent data being None
for key, value in data.items():
try:
setattr(self, key, value)
except AttributeError:
pass
def retrieve(self, *, invalidate_cache: bool = False) -> bool:
"""Retrieve object data from the cache or endpoint (setting cache).
Uses a cache to avoid unnecessary queries to the endpoint. Set
invalidate_cache to force a new query.
:param invalidate_cache: Force querying the endpoint and overwriting the
cache.
:return: True if retrieval was successful, False otherwise.
"""
result = True # For a cache hit
if invalidate_cache or not self.cache_retrieve():
data = self._get_data_from_endpoint()
result = bool(data)
self.unserialize(data)
self.save()
return result
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.