A very simple async Response cache for FastAPI

Hello world! How’s it going out there? Monkey pox scars healing OK? Great.

My most recent project is pythondocs.xyz – real time interactive search of Python documentation.

If you haven’t checked it out yet, please take a look and let me know what you think over here.

The site is built with FastAPI and I wanted to make it as fast as possible. In particular, I wanted the home page to load almost instantly. The home page is constructed from a couple of database queries and I realised I could reduce load times by building the page once then caching it for future visitors.

But how?

Decorated service

Here’s the solution I came up with, with no external dependencies.

To start, some imports:

import asyncio
from functools import wraps

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

The main course is where you find the meat:

def cache_response(func):
    """
    Decorator that caches the response of a FastAPI async function.

    Example:
    ```
        app = FastAPI()

        @app.get("/")
        @cache_response
        async def example():
            return {"message": "Hello World"}
    ```
    """
    response = None

    @wraps(func)
    async def wrapper(*args, **kwargs):
        nonlocal response
        if not response:
            response = await func(*args, **kwargs)
        return response

    return wrapper

And what’s for dessert? Oh, my favorite: objects! They’ll come in handy for the examples below…

app = FastAPI()
templates = Jinja2Templates(directory="templates")

What’s going on?

The idea is that you insert the @cache_response decorator between your async function and its FastAPI path operation decorator.

Here’s a simple example:

@app.get("/json/")
@cache_response
async def json_example():
    await asyncio.sleep(2)
    return {"message": "Hello World"}

When someone navigates to /json/, the json_example function is passed into the wrapper as func.

On the first visit, the wrapper awaits func, stores its response for next time, and returns the response.

On subsequent visits, the response is returned directly, func is never called, and whatever expensive operations it contains don’t slow things down.

Template tantrum

The decorator also works with functions that have parameters, like this Jinja template route:

@app.get("/", response_class=HTMLResponse)
@cache_response
async def home_page(request: Request):
    await asyncio.sleep(2)
    return templates.TemplateResponse("hello_world.html", context={"request": request})

But beware that the first response is always stored and reused, even if called with different arguments!

Is this a terrible idea?

I don’t think it’s, like, the worst idea ever. It’s not like writing your Social Security Number on your front door.

But this is computers and there’s a million ways of doing everything, with various strengths and weaknesses.

I think these are the main downsides of this very simple approach:

  • The cached data is stored in-process and so won’t be shared with other workers.
  • There’s no way of clearing the cache (other than restarting Python).
  • A response isn’t cached until the first call completes. Additional requests received before this happens will still trigger your expensive function.

Here are some alternative approaches, which may work better for your case:

  • Cache intermediate steps in your function (rather than the whole thing), perhaps with functools.cache.
  • Use a dedicated caching service, like Redis, or a database.
  • Cache static assets with a Content Delivery Network, like Cloudflare.

Thanks for reading

More Python stuff coming soon.

Let me know if you have any comments on this post or ideas for the future!

The full code for this post is here on GitHub.

2 thoughts on “A very simple async Response cache for FastAPI

  1. Can’t use pickle dumps response.
    AttributeError: Can’t pickle local object ‘Jinja2Templates._create_env..url_for’

  2. I’ve slightly edited the code to check the freshness of the cache.
    In this example the cache is valid for 10 seconds.
    used timedelta from datetime for it.

    @wraps(func)
    async def wrapper(*args, **kwargs):
    nonlocal response
    nonlocal last_updated

    print(response)
    if not last_updated:
    last_updated = datetime.now()
    response = await func(*args, **kwargs)
    return response

    now = datetime.now()
    if now – timedelta(seconds=10) > last_updated:
    last_updated = datetime.now()
    response = await func(*args, **kwargs)
    return response

    return response

Leave a Reply

Your email address will not be published. Required fields are marked *