Кэширование в FastAPI: Redis-обёртка, которую можно повторно использовать
Кэш в API-сервисе — это та штука, которая кажется простой, пока ты её сам не написал. «Положи результат в Redis на 5 минут». Звучит элементарно. А потом начинается: ключи, инвалидация, сериализация, обработка ошибок, разные типы данных, разные TTL, и через год это уже клубок утилит, разбросанных по проекту.
Я расскажу про обёртку, которая у нас работает в трёх FastAPI-сервисах. Не библиотека, а скорее готовый рецепт: один файл, который ты копируешь в проект и адаптируешь под себя. Главное — что она решает реальные проблемы, а не выглядит красиво.
Что должно быть в обёртке
Минимальный список требований, которые мы выработали:
- Один способ кэшировать функцию через декоратор.
- Поддержка async-функций.
- Серриализация через JSON (быстро) с возможностью сменить на pickle (для сложных типов).
- Адекватные ключи: имя функции + аргументы.
- Игнорирование «технических» аргументов (например,
session). - Инвалидация по тегам: «выкинь весь кэш, связанный с user=123».
- Не падать, если Redis недоступен. Пусть просто прокладывает запрос мимо кэша.
Почти всегда люди начинают с пунктов 1–3, через месяц добавляют 4, через полгода — 6, и в итоге у них своя версия того же самого, только криво.
Базовый клиент Redis
Используем redis.asyncio (он же redis-py async-режим, после слияния aioredis):
from redis.asyncio import Redis, ConnectionPool
pool = ConnectionPool.from_url(
"redis://redis:6379/0",
max_connections=50,
decode_responses=True,
)
def get_redis() -> Redis:
return Redis(connection_pool=pool)Один пул на процесс, инстанс Redis — лёгкий объект, его можно создавать на каждый запрос. decode_responses=True сразу даёт строки вместо bytes — удобно, пока ты не работаешь с бинарными значениями.
Сериализация и ключи
Берём JSON по умолчанию: быстро, читабельно, межязыковая совместимость. Для нестандартных типов (datetime, Decimal, UUID) добавляем кастомный энкодер:
import json
from datetime import datetime
from decimal import Decimal
from uuid import UUID
class JSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Decimal):
return str(obj)
if isinstance(obj, UUID):
return str(obj)
return super().default(obj)
def dumps(value) -> str:
return json.dumps(value, cls=JSONEncoder, separators=(",", ":"))
def loads(value: str):
return json.loads(value)Для ключа собираем стабильную строку из имени функции и аргументов. Стабильность важна: один и тот же набор аргументов должен давать один ключ независимо от порядка kwargs.
import hashlib
def _build_key(prefix: str, func_name: str, args: tuple, kwargs: dict, ignore: set[str]) -> str:
payload = {
"args": args,
"kwargs": {k: v for k, v in sorted(kwargs.items()) if k not in ignore},
}
raw = dumps(payload)
digest = hashlib.sha1(raw.encode()).hexdigest()[:16]
return f"{prefix}:{func_name}:{digest}"SHA-1 — не для криптостойкости, а для длины и распределения. Берём первые 16 символов: ключи короткие, читаемые, коллизии практически невозможны на нашей нагрузке.
Декоратор: ядро обёртки
Код довольно компактный. Вот живой вариант, который мы используем:
import functools
import logging
from typing import Callable, Iterable
log = logging.getLogger(__name__)
def cached(
*,
ttl: int = 60,
prefix: str = "cache",
ignore_args: Iterable[str] = ("session", "request", "background_tasks"),
tags: Callable[..., list[str]] | None = None,
):
ignore_set = set(ignore_args)
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
redis = get_redis()
key = _build_key(prefix, func.__qualname__, args, kwargs, ignore_set)
try:
cached_value = await redis.get(key)
except Exception as exc:
log.warning("cache get failed: %s", exc)
cached_value = None
if cached_value is not None:
return loads(cached_value)
result = await func(*args, **kwargs)
try:
pipe = redis.pipeline()
pipe.set(key, dumps(result), ex=ttl)
if tags:
for tag in tags(*args, **kwargs):
pipe.sadd(f"{prefix}:tag:{tag}", key)
pipe.expire(f"{prefix}:tag:{tag}", ttl + 60)
await pipe.execute()
except Exception as exc:
log.warning("cache set failed: %s", exc)
return result
return wrapper
return decoratorЧто важно:
- Любая ошибка Redis ловится и логируется. Запрос всегда возвращается, даже если Redis лёг. Это критично — недоступный кэш не должен ронять бизнес-логику.
- Параметр
ignore_argsисключает «технические» аргументы из ключа. Без этого в ключ попадает адресAsyncSession, и ты получаешь cache miss на каждый запрос. tags— функция, возвращающая список тегов по аргументам. На каждый тег ведём set ключей, которые с ним связаны.
Использование
@cached(
ttl=300,
prefix="users",
tags=lambda user_id, **_: [f"user:{user_id}"],
)
async def get_user_profile(user_id: int, session: AsyncSession) -> dict:
user = await session.get(User, user_id)
return {"id": user.id, "name": user.name, "email": user.email}В ключ попадёт только user_id, потому что session в ignore_args. Тег user:123 позволит позже инвалидировать всё связанное с этим пользователем.
Инвалидация
Отдельная функция для сброса по тегам:
async def invalidate_tags(*tags: str, prefix: str = "cache") -> int:
redis = get_redis()
total = 0
for tag in tags:
tag_key = f"{prefix}:tag:{tag}"
keys = await redis.smembers(tag_key)
if keys:
await redis.delete(*keys)
total += len(keys)
await redis.delete(tag_key)
return total
# где-то в эндпоинте после изменения пользователя
await invalidate_tags(f"user:{user.id}", prefix="users")Логика прямолинейная: достаём из set все ключи, связанные с тегом, удаляем их и сам set. На больших базах ключей это безопасно — мы не делаем SCAN и не блокируем Redis.
Подводные камни, которые мы прошли
Thundering herd. Когда популярный ключ протух, все запросы одновременно идут в БД, считают и пишут одно и то же. На больших нагрузках это убивает базу. Решение — добавить блокировку или SETNX-флаг «считаю, не лезьте». Самый простой вариант: при miss один поток ставит «building» флаг, остальные кратко спят и читают результат.
Cache stampede на старте. После рестарта пода Redis пустой (если он у тебя локальный) или просто кэш холодный, и пик нагрузки идёт прямо на БД. Если важно — разогревай кэш на старте через lifespan.
Большие значения. Если кэшируешь объект на 500 КБ — Redis это съест, но будет неловко. Лучше кэшировать кусочки, а не гигантские агрегаты целиком.
Сериализация изменилась. Сменил структуру ответа функции — старые значения в кэше становятся несовместимы. Версионируй prefix: users.v2. На рестарте новые ключи, старые сами протухнут.
TTL у тегов. Если тег живёт меньше, чем закэшированные ключи, ты не сможешь корректно инвалидировать. Поэтому TTL тега всегда чуть больше TTL значений.
Что не делать
- Не используй pickle, если данные могут быть прочитаны другим сервисом или другой версией Python. JSON надёжнее.
- Не кэшируй на основе
request: получишь утечку приватных данных между пользователями. - Не делай TTL
Noneи не полагайся на ручную инвалидацию. Любой кэш должен иметь верхнюю границу жизни. - Не пихай весь кэш в
db=0. Разные кэши — разные БД Redis или хотя бы разные префиксы. Это упрощает диагностику.
Сухой итог
Эта обёртка — не серебряная пуля и не замена нормальному кэширующему слою. Это рабочий минимум, который закрывает 90% случаев в прикладном коде. Декоратор, тегированная инвалидация, отказоустойчивость к Redis. С неё хорошо начинать; когда упрёшься в её ограничения — будешь знать, какую конкретно фичу добавлять.
Главное правило кэша: он должен ускорять, но не быть источником непредсказуемых ошибок. Если кэш падает, сервис должен продолжать работать. Это первая проверка, которую стоит проводить на любом кэширующем коде, прежде чем катить в прод.