Cache in asynchronous Python applications

Foreword

Modern Internet would be impossible without caching. It is present literally everywhere in the web app: browser keeps its own cache to speed up page loading, CDN helps to serve static files faster, databases use buffer cache to optimize i/o operations on high rate queries. And applications are no exception. Unfortunately, most applications don’t use caching as much as they could. This is because developers typically use caching as a last resort to speed up a slow application. Caching also adds some overhead to an application: keeping the cache consistent and invalidate it at the right time is not an easy task.

In the article, we will talk about caching techniques in asynchronous Python applications. First, we’ll try to implement it on our own and then will take a look at third-party libraries.

Why asynchronous python, exactly? Regular python already has a lot of production-ready third-party libraries. Also, async python gives us some interesting possibilities, for example, we can run multiple coroutines to manage the cache.

Simple cache implementation

And so let’s begin. Eventually, you want to use cache in your application. The naive implementation can look something like this:

cache = {}
KEY_TEMPLATE = "user_account-{user_id}-{account_id}"

async def get_user_account(user: User, account_id: int):
    key = KEY_TEMPLATE.format(user_id=user.id, account_id=account_id)
    if key not in cache:
        cache[key] = await _get_user_account(user, account_id)
    return cache[key]

Quite a simple and popular solution, but it has some cons:

Some of them are related to the code maintainability:

1. Sometimes you need to bypass the cache and get the fresh data based on some conditions. To do this you can add a parameter use_cache=False to the function or have two functions: get_user_account_with_cache and get_user_account. However, this is not very convenient to support.
2. You need to carefully write your tests. Somehow you need to test that the cache is actually working correctly. That means you need to test for several scenarios — cache hit, cache miss, etc…

Another important problem is that there is no cache size limit. As your cache grows, so does memory consumption. This is a memory leak because we never discard any items we put in the cache. Therefore, we need “smart” storage with one of the cache replacement algorithms — LRU, MRU, LFU, TTL. Of course, you can write a bunch of code and implement smart in-memory storage yourself or just use a third-party library, for example, async-lru. However, when using Kubernetes and having several pods with your application (that can possibly restart multiple times per day) such cache will be inefficient.

Modern key-value databases can limit the cache out of the box. A good example of such storage is Redis, which not only allows you to specify TTL (“time to live”) for the key but also has different preemptive mechanisms that are enabled by simple settings.

Inevitably, our code will grow to something like this:

TEMPLATE = "user_account-{user_id}-{account_id}"
CACHE_TTL_SECONDS = 24 * 60 * 60 # 24 hours
cache = CacheStorage()

async def get_user_account(user: User, account_id: int):
    key = TEMPLATE.format(user_id=user.id, account_id=account_id)
    account = await cache.get(key)
    if account:
        return account
    account = await _get_user_account(user, account_id)
    await cache.set(key, account, ttl=CACHE_TTL_SECONDS)
    return account

This will work fine as long as account is just a string, or another primitive type, but what if account is an object. In that case, you have to serialize and deserialize it. Usually, either json or pickle are used for this. But let’s say that this is already implemented at the CacheStorage level.

While this implementation works it’s still a lot of boilerplate code. Imagine, that every time you want to use cache you need to write (or copy-paste) something like in the example above. In other words, it scales poorly. So… let’s try to solve this issue with a… Decorator!

# never use it
import asyncio
import json
from functools import wraps
CACHE_TTL_HOURS = 24
_cache = CacheStorage()

def cache(function):
    @wraps(function)
    async def _wrapped(*args, **kwargs):
        cache_key = function.__module__ + function.__name__ + json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True)
        result = await _cache.get(cache_key)
        if result:
            return result
        result = await function(*args, **kwargs)
        asyncio.create_task(_cache.set(cache_key, result, ttl=CACHE_TTL_HOURS))
        return result
    return _wrapped

This naive implementation solves many problems described above:

• It is a single implementation that is easy to change (no code duplication).

But we still have problems:

• the key is ugly because it essentially contains json
• the key depends on the function call — the arguments can be passed as positional or as named
• not all objects can be serialized to json out of the box
• we cannot control the key format and TTL.

WARNING! do not use “hash” function

do not use `hash(obj)` as a cache key — it is not safe because the hash return value can change between different python processes — calling hash function on the same object in two different python processes gives you different results. Using other hashes, such as md5 or sha, as keys are not recommended either, as it makes troubleshooting difficult as it is not human readable .

Slightly more complex implementation

Let’s extend the decorator and allow a user to customize a key format by providing a special key function. This function will take function call arguments and return a key:

import asyncio
from functools import wraps
from datetime import timedelta

def cache(key_function: Callable, ttl: timedelta):
    def _decorator(function):
        @wraps(function)
        async def _function(*args, **kwargs):
            cache_key = key_function(*args, **kwargs)
            result = await _cache.get(cache_key)
            if result:
                return result
            result = await function(*args, **kwargs)
            asyncio.create_task(_cache.set(cache_key, result, ttl=ttl.total_seconds()))
            return result
        return _function
    return _decorator


def _key_function(user, account_id):
    return f"user-account-{user.id}-{account_id}"


@cache(_key_function, timedelta(hours=3))
async def get_user_account(*, user: User, account_id: int):
    ...

Such a decorator is more predictable and easier to use, but you would have to adapt the function signature to make it easier to describe the key formatting function. Plus, there are a number of questions. How to call the function without cache? Is it possible to understand whether the value was obtained from the cache or by calling a decorated function? How to forcibly invalidate the cache? These complexities and questions lead to the fact that it would be much easier to use ready-made implementations.

Cashing Lib

Of course, many developers will prefer to use open-source packages for the functionality described above and this is a smart decision. However, there are not many libraries that are sharpened for working with asynchronous Python. I also faced the same problem.

The most popular cache management library is aiocache. It provides many possibilities:

• support for different backends: redis, memory, memcached
• different serialization mechanisms: pickle, json
• decorator with approximately the same functionality as described earlier
• plugin mechanism

Unfortunately, this library does not answer the questions addressed earlier, namely:

• ability to call the function without the cache
• API to understand if the result was taken from the cache or not
• built-in mechanisms of forced invalidation

Own implementation: meet the Cashews!

All the shortcomings of existing libraries and additional needs prompted me to write my own library. That’s how it looks like:

from cashews import cache
cache.setup("redis://") 

@cache(ttl="3h", key="user:{user.id}:{account_id}")
async def get_user_account(user: User, account_id: int) -> Account: 
    ...

I would like to show right away how this library helps with the problems I mentioned above:

1) ability to call the function without the cache, disable it dynamically

Usage sample: we have a web server and we have endpoints that have a cache. We want to disable the cache if the client set Cache-Control header:

from cashews import cache

@app.middleware("http")
async def disable_cache_for_no_store(request: Request, call_next):
    if request.method.lower() != "get":
        return await call_next(request)
    if request.headers.get("Cache-Control") in ("no-store", "no-cache"):
        with cache.disabling("get", "set"):
            return await call_next(request)
    return await call_next(request)

Here the context manager cache.disabling(“get”, “set”) disables get and set operations on the cache. Thus, the cache is not accessed if the request has a header Cache-Control equal no-cache.

2) API to understand if the result was from the cache or not

Usage sample: we have a web application and we want to add to the headers an indication that the cache was used for the request result. Let’s make middleware:

from cashews import cache

@app.middleware("http")
async def add_from_cache_headers(request: Request, call_next):
    with cache.detect as detector:
        response = await call_next(request)
        if request.method.lower() != "get":
            return response
        if detector.calls:
            response.headers["X-From-Cache-keys"] = ";".join(detector.calls.keys())
    return response

Here the context manager cache.detect tracks successful cache calls and puts them into the detector. And then we check, if something got into it, it means that we had cache “hit”. You can get a list of keys from detector and add it to our custom header.

3) built-in mechanisms of forced invalidation.

Usage sample: we have a CRUD API On read we have a cache, but it needs to be invalidated after operations create, update or delete

from cashews import cache

@app.get("/friends")
@cache(ttl="10h")
async def get_fiends(user: User = Depends(get_current_user)):
    ...


@app.post("/friends")
@cache.invalidate(get_fiends)
async def create_friend(user: User = Depends(get_current_user)):
    ...

The idea is that if create_friend is called successfully, the library itself will generate the key that is used for the cache and delete the key/keys.

Other features of Cashews

And I wanted to tell you a little bit about the cool features of this library:

1) At the moment, the library supports the following storages (“backends”):
cache.setup(“mem://”, size=1000) # inmemory (LRU + ttl)

cache.setup(“mem://”, size=1000) # inmemory (LRU + ttl)
cache.setup(“disk://?directory=/tmp/cache”) # use filesystem
cache.setup(“redis://redis_host/0”) # use redis

2) Client-side cache. In general, these backends would be enough, but there is another one that appeared by a new feature of Redis 6. A client-side cache — is a mechanism that Redis provides to keep the local copy of the cache consistent. It’s very simple. In writing we put the key-value pair in Redis and in memory, also keep a separate connection that is subscribed for key invalidation events from Redis. When the event is received, we erase the value from our local cache by the key. This mechanism makes it possible to have a very fast cache (because it’s in memory) that does not get cold after restarting the application.

cache.setup(“redis://0.0.0.0”, client_side=True)

3) Using different backends depending on the key prefix

cache.setup(“redis://redis”)
cache.setup(“mem://”, prefix=”users:”)
await cache.get(“key”) # use redis backend
await cache.get(“users:1”) # use memory backend

4) Different “strategies” for using the cache:

@cache(ttl=”5h”)– the most common cache, as we described at the beginning of this article.

@cache.hit(ttl=”1d”, cache_hits=500, update_after=100)– in addition to expiration by TTL, expiration by the number of hits is added. With update_after the cache will be updated in the background after the specified number of hits and reset the counter. Can be used for Cache Stampede prevention.

@cache.early(ttl=”10m”, early_ttl=”7m”) — cache which besides normal TTL has early expiration TTL, after which the cache will be updated in the background and reset the TTL. Can be used for Cache Stampede prevention.

@cache.failover(ttl=”2h”, exceptions=(ValueError, MyException)) — this decorator stores the result of the response almost always, however, if the function call ends with specified exceptions, it will return the previously saved value. This is especially useful for external calls when the service can be unavailable from time to time, but we would like to “smooth out” its crashes.

5) General set of utilities-decorators for microservices like Circuit breaker, Shared Lock, Rate limit

In general, if you want to use the cache efficiently, use cashews ☺

Links:

cashews — https://github.com/Krukov/cashews

aiocache — https://github.com/aio-libs/aiocache