Base Cache Strategy
To implement a custom cache strategy, you need to create a new class that extends the CacheStrategy
class and implement the abstract methods.
See a naive implementation of a memcached cache strategy below:
from io import BytesIO
from pymemcache.client import base
import pandas as pd
import ckanext.charts.config as config
from ckanext.charts.cache import CacheStrategy
class MemcachedCache(CacheStrategy):
"""Cache data to Memcached"""
def __init__(self):
self.client = base.Client(('localhost', 11211))
def get_data(self, key: str) -> pd.DataFrame | None:
"""Return data from cache if exists"""
try:
raw_data = self.client.get(key)
if not raw_data:
return None
return pd.read_csv(BytesIO(raw_data))
except Exception:
log.exception(f"Failed to get data for key: {key}")
return None
def set_data(self, key: str, data: pd.DataFrame):
"""Serialize data and save to Memcached"""
cache_ttl = config.get_memcached_cache_ttl()
try:
serialized_data = data.to_csv(index=False).encode('utf-8')
self.client.set(key, serialized_data, expire=cache_ttl)
except Exception:
log.exception(f"Failed to save data to Memcached for key: {key}")
def invalidate(self, key: str):
"""Invalidate cache by key"""
try:
self.client.delete(key)
except Exception:
log.exception(f"Failed to invalidate cache for key: {key}")
Bases: ABC
Cache strategy interface.
Defines the abstracts methods for cache strategies.
get_data(key)
abstractmethod
Return data from cache if exists.
PARAMETER | DESCRIPTION |
---|---|
key
|
The cache key to retrieve the data.
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
DataFrame | None
|
The data if exists, otherwise None. |
invalidate(key)
abstractmethod
Invalidate cache by key.
PARAMETER | DESCRIPTION |
---|---|
key
|
The cache key to invalidate.
TYPE:
|
set_data(key, data)
abstractmethod
Store data to cache.
PARAMETER | DESCRIPTION |
---|---|
key
|
The cache key to store the data.
TYPE:
|
data
|
The data to be stored.
TYPE:
|