FastAPI with Redis Caching
I have been working on a FastAPI project lately and I needed to cache fastapi response and function results. This post describes how to use fastapi with redis caching.
Let’s first create a Python virtual environment called venv
, activate it and install the requirements.
python3 -m venv venv
source ./venv/bin/activate
pip install fastapi==0.89.1 redis==4.4.2 fastapi-cache2==0.2.0 "uvicorn[standard]"
A minimal FastAPI application. Create a main.py
alongside your venv
and save the below content.
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def index():
return {"message": "Hello, world!"}
Then run the app with uvicorn in development mode.
uvicorn main:app --reload
Or you can create a simple bash script run.sh
to run it, it will save you time when developing.
#!/bin/bash
uvicorn main:app --reload
The chmod +x
command will make the file run.sh
an executable so you can run it like ./run.sh
.
chmod +x ./run.sh
./run.sh
With uvicorn server running, visit http://127.0.0.1:8000 and if all goes well, you will be greeted with json response with a “Hello, world!” message on it.
And the interactive API documentation provided by swagger ui. Visit http://127.0.0.1:8000/docs
Below is how to connect FastAPI with Redis in an asynchronous way. See Asyncio Redis for more info. If you notice that explicitly getting and setting values and encoding/decoding it can become tedious so let’s find a python library to simplify this for us.
import json
from fastapi import FastAPI
from redis import asyncio as aioredis
app = FastAPI()
redis = aioredis.from_url("redis://localhost", encoding="utf8", decode_responses=True)
@app.get("/")
async def index():
key: str = 'my-key'
value = await redis.get(key)
if value is None:
value = {"message": "Hello, world!"}
await redis.set(key, json.dumps(value), 5*60)
return json.loads(value)
Let’s use fastapi-cache to automatically cache function response and can be setup to use redis backend too.
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis
app = FastAPI()
@app.on_event("startup")
async def startup():
redis = aioredis.from_url("redis://localhost", encoding="utf8", decode_responses=True)
FastAPICache.init(RedisBackend(redis), prefix="api:cache")
@app.get("/")
@cache(expire=60*5)
async def index():
return {"message": "Hello, world!"}
Now we can just use the cache
decorator imported from fastapi-cache
to cache function results!
The cache
decorator can receive optional parameters, here I set expire
to 5 * 60 which will expire
the cache content after 300 seconds or 5 minutes. Check for more cache decorator optional parameters.
To see if it really cached the result, let’s check via redis-cli
.
Yes, indeed! So it can cache json responses, let try to cache html response!
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis
from fastapi.responses import HTMLResponse
app = FastAPI()
@app.on_event("startup")
async def startup():
redis = aioredis.from_url("redis://localhost", encoding="utf8", decode_responses=True)
FastAPICache.init(RedisBackend(redis), prefix="api:cache")
@app.get("/")
@cache(expire=60*5)
async def index():
return {"message": "Hello, world!"}
@app.get("/about", response_class=HTMLResponse)
@cache(expire=60*60)
async def about():
return '<p style="color: blue;">This is the about page.</p>'
Running the above will result like below.
And the cached html response in redis via redis-cli
.
Hmm I cached both the index
and about
routes but looking at the cache keys in redis, I cant determine
which keys it belongs to. I have a requiment where I will get all the cache keys and display it to an html list.
The user then can select a particular redis key to delete so that a fresh/new content can be generated from the route or view.
Thankfully fastapi-cache
allows us to use a custom redis key builder!
from urllib.parse import urlparse
from typing import Optional
from starlette.requests import Request
from starlette.responses import Response
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis
from fastapi.responses import HTMLResponse
app = FastAPI()
def cache_key_builder(
func,
namespace: Optional[str] = "",
request: Request = None,
response: Response = None,
*args,
**kwargs
):
prefix = FastAPICache.get_prefix()
if request:
parsed = urlparse(str(request.url))
cache_key = f"{prefix}:{namespace}:{func.__module__}:{func.__name__}:{parsed.path}"
else:
cache_key = f"{prefix}:{namespace}:{func.__module__}:{func.__name__}:{args}:{kwargs}"
return cache_key
@app.on_event("startup")
async def startup():
redis = aioredis.from_url("redis://localhost", encoding="utf8", decode_responses=True)
FastAPICache.init(RedisBackend(redis), prefix="api:cache")
@app.get("/")
@cache(expire=60*5)
async def index():
return {"message": "Hello, world!"}
@app.get("/about", response_class=HTMLResponse)
@cache(expire=60*60, key_builder=cache_key_builder)
async def about():
return '<p style="color: blue;">This is the about page.</p>'
We are now using an optional parameter key_builder
with the value of our custom cache_key_builder
in the cache
decorator.
I applied the cache decorator on the about
page and it should now have a custom redis key on it.
Let’s check it via redis-cli
.
And that’s a wrap! FastAPI + Redis = 🚀