Django cache for APIs with invalidation

Photo by Jp Valery on Unsplash

Django cache for APIs with invalidation

APIs especially for dashboards are predominantly read-heavy. To make for a user friendly experience, we should reduce response times of the APIs.

The tried and trusted way uses a distributed cache-aside strategy or something more exotic. The expensive computation is done once, storing the serialized result on a lookup key which the API references to return the serialized data. Typically the key would be of the API's url segmented by a user.

With DjangoRESTFramework we can leverage decorators

  • @cache_page(time-in-seconds)

  • @vary_on_header(user-identifier)

on the GET API views to cache-aside content.

If the data is static or your use case tolerates some staleness, you can stop reading here.

But if the content changes periodically you could serve stale data before time-in-seconds expires.

If you know the set of urls that need to be invalidated, you could just slap them into the respective post save/delete handlers and you'd be done, albeit with fragility.

However if you can't know all the urls because the user-identifier is not present in headers/cookies or pagination parameters get appended to the url, we would need to

  • generate a dynamic prefix for all possible urls to cache content against

  • use a caching backend that allow for a regex key space search for invalidation

The latter could be resolved via django-redis.

The former via the decorator* below where the dynamic urls and generated prefix can leverage the decorated methods' args/kwargs references to compute content at declaration.

def cached_api_response(
    Cache response of the decorated method
    `key_suffix_or_callable: callable(*args, *kwargs) | str`
    - ex: `key_suffix_or_callable: lambda _, request: 'request.get_full_path()`
    `user_identifier_or_callable: callable(*args, *kwargs) | str`
    - ex: `user_identifier_or_callable: lambda obj, _: obj.user_id`
    `timeout: int` TTL for cache in seconds
    def decorator(func):
        def func_wrapper(*args, **kwargs):
                key = (
                    key_suffix_or_callable(*args, **kwargs)
                    if callable(key_suffix_or_callable) 
                    else key_suffix_or_callable
                user_id = (
                    user_identifier_or_callable(*args, **kwargs) 
                    if callable(agent_id) else agent_id
                prefixed_key = f'foobar_{user_id}_{key}'
                if (cached_data := cache.get(prefixed_key)):
                    return cached_data
                result = func(*args, **kwargs)
                cache.set(prefixed_key, result, timeout=ttl)
            except Exception as e:
                    'Error %s while caching %s', e, func.__name__,),
                result = func(*args, **kwargs)
            return result
        return func_wrapper
    return decorator

\this generalized, concise code wasn't crafted by myself*