Cache#
A common way to optimize applications is by using caching, which involves storing the outputs of certain processes for a set period. For FastAPI, the fastapi_cache library provides a convenient way to implement caching, simplifying the process significantly.
from random import random
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from fastapi_cache.backends.redis import RedisBackend
from httpx import AsyncClient
from redis import asyncio as aioredis
For testing, we need a cache backend. Here, we’ll use Redis in a Docker container—the following cell creates this container.
import docker
docker_client = docker.from_env()
container = docker_client.containers.run(
image="redis:7.4.0",
name="fastapi_chache_test_redis",
detach=True,
remove=True,
ports={6379: 6380}
)
Let’s try gotten redis backend.
redis_client = aioredis.from_url("redis://localhost:6380")
await redis_client.ping()
True
The following line is crucial—it initializes FastAPICache. It’s typical to use this initialization within the lifespan function of your application.
FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")
Note: Don’t forget to stop the Docker container with Redis after you’re done.
container.stop()
Check cache#
Caching isn’t magic; you can find your cached records in the backend you’re using.
The following example defines a program that returns a random value each time you access the endpoint, requiring you to specify a user_id.
app = FastAPI()
@app.get("/{id}")
@cache(expire=600)
def index(id: int):
return random()
The following code executes the application and makes requests using different IDs, with each ID being requested twice.
async with AsyncClient(app=app, base_url="http://test") as ac:
content10_1 = (await ac.get("/10")).content
content10_2 = (await ac.get("/10")).content
content20_1 = (await ac.get("/20")).content
content20_2 = (await ac.get("/20")).content
display(content10_1)
display(content10_2)
display(content20_1)
display(content20_2)
b'0.45806010585100077'
b'0.45806010585100077'
b'0.8354224226556365'
b'0.8354224226556365'
As a result, requests with the same IDs return the same numbers, demonstrating that caching is working.
Let’s check the keys available in the experimental Redis database and the values corresponding to them.
keys = await redis_client.keys("*")
values = await redis_client.mget(keys)
for key, value in zip(keys, values):
print(f"{key.decode('utf-8')}: {value.decode('utf-8')}")
fastapi-cache::1ef05260a7f5a4b566cd6a44d5147ea7: 0.8354224226556365
fastapi-cache::83ad8bee65395e62425345c322d43f64: 0.45806010585100077
The values retrieved from Redis match the values we received from the API responses.
Query params#
You can also use query parameters with fastapi_cache. Each unique combination of query parameters will have its own cached variable.
The following API defines an endpoint that uses query parameters.
app = FastAPI()
@app.get("/")
@cache(expire=600)
def index(id: int):
return random()
Now let’s try passing different query parameters to the API.
async with AsyncClient(app=app, base_url="http://test") as ac:
content_id50_1 = (await ac.get("/?id=50")).content
content_id50_2 = (await ac.get("/?id=50")).content
content_id60_1 = (await ac.get("/?id=60")).content
content_id60_2 = (await ac.get("/?id=60")).content
display(content_id50_1)
display(content_id50_2)
display(content_id60_1)
display(content_id60_2)
b'0.33010852478769437'
b'0.33010852478769437'
b'0.41410719388169837'
b'0.41410719388169837'
Different query parameters will produce different results. However, identical sets of query parameters will return the same values due to caching.
Wrong backend#
It’s interesting to note that if you define an incorrect backend for your application, everything will work fine except for caching, of course.
The following cell defines an application that first resets FastAPICache and then redefines it with a non-existent Redis backend.
wrong_client = aioredis.from_url("redis://this_host_doesnt_exist:7777")
FastAPICache.reset()
FastAPICache.init(RedisBackend(wrong_client), prefix="fastapi-cache")
app = FastAPI()
@app.get("/")
@cache(expire=600)
def index():
return random()
The following cell shows that appliactoin actually answers requests but caching doesn’t work.
async with AsyncClient(app=app, base_url="http://test") as ac:
print((await ac.get("/")).content)
print((await ac.get("/")).content)
b'0.37765920837730704'
b'0.05150428134172702'
After this example, you’ll need to revert the FastAPICache initialization to ensure that other examples work correctly.
FastAPICache.reset()
FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")
Key builder#
A special function that generates keys to be used in Redis for caching. You can get default key_builder by using FastAPICache.get_key_builder method of the initialised FastAPICache.
The following cell demonstrates how to retrieve the default key_builder and even calls it with an incomplete set of arguments.
key_builder = FastAPICache.get_key_builder()
def test_function():
pass
key_builder(func=test_function, args=[], kwargs={})
':f0a9508ba2bac36e3742d56a5c0859cb'
As a result, a random hash is generated.
Exceptions caching#
fastapi_cache does not provide caching of exceptions. Therefore, if your code throws an exception, each subsequent call to the same endpoint will re-execute all code leading up to the exception.
The following code provides an application that throws an exception every time it’s called. The exception message is generated randomly.
app = FastAPI()
@app.get("/")
@cache(expire=600)
def index():
raise HTTPException(500, str(random()))
Each call to the API results in an error with a different message.
async with AsyncClient(app=app, base_url="http://test") as ac:
print((await ac.get("/?id=50")).content)
b'{"detail":"0.2278803117705277"}'
JSONResponse as exception#
As possible solution you can reproduce outputs of the fastapi.exceptions.HTTPException using fastapi.responses.JSONResponse.
The following cell demonstrates the creation of an application that generates a JSONResponse with a status_code=500 and a random message under the detail key for each response.
app = FastAPI()
@app.get("/")
@cache(expire=600)
def index():
data = {"detail": str(random())}
return JSONResponse(content=data, status_code=500)
Now let’s make some requests to the created application.
async with AsyncClient(app=app, base_url="http://test") as ac:
response1 = await ac.get("/")
response2 = await ac.get("/")
print("=========response 1=========")
print("Status code", response1.status_code)
print("Content", response1.content)
print("Headers", response1.headers)
print("=========response 2=========")
print("Status code", response2.status_code)
print("Content", response2.content)
print("Headers", response2.headers)
=========response 1=========
Status code 500
Content b'{"detail":"0.35714545508611095"}'
Headers Headers({'content-length': '32', 'content-type': 'application/json'})
=========response 2=========
Status code 200
Content b'{"detail":"0.35714545508611095"}'
Headers Headers({'content-length': '32', 'content-type': 'application/json', 'cache-control': 'max-age=600', 'etag': 'W/3775337072360237072', 'x-fastapi-cache': 'HIT'})
Actually content was cached but not status code!