Опирается на правила: R-CACHE-CFG-1 … R-CACHE-CFG-5 и R-CACHE-CFG-X1 … R-CACHE-CFG-X4 из Caching Style Guide → раздел 2. Конфигурация.

Важно знать

  • Redis в проде через redis.asyncio, не dict/cachetools. В multi-instance каждый процесс имел бы свой локальный кеш.
  • JSON-сериализация через json.dumps/json.loads (или orjson). pickle — security risk: RCE при десериализации недоверенных данных.
  • Per-cache TTL — каждый namespace имеет explicit срок; один глобальный TTL — компромисс ни для кого.
  • pydantic-settings для cache settings: TTL меняется в .env/application.yml, не в коде.
  • Кеш-порт — Protocol в core/, реализация в adapters/out/cache/. Хендлер зависит от абстракции, не от redis.asyncio напрямую.
  • В тестах — fakeredis или Testcontainers Redis. Мок cache-порта теряет поведение TTL и eviction.
  • aiocache-декораторы — удобная альтернатива явному cache-aside для простых read-методов; serializer = JsonSerializer, не PickleSerializer.
  • Явная инициализация клиента — Redis.from_url(url, decode_responses=True); без decode_responses=True получаете bytes вместо str.

Redis вместо dict в проде

R-CACHE-CFG-1: production backend — Redis.

Почему не dict / lru_cache / cachetools:

  • Multi-instance. Gunicorn с 4 воркерами или 10 реплик в K8s = 10 изолированных dict. customer_id=42 попадает в разные процессы и читает разные значения.
  • Invalidation race. Write в одном процессе не уведомляет другие — старое значение живёт до TTL.
  • Persistence. Redis с RDB/AOF переживает restart. dict теряет всё.
  • Observability. Redis-side метрики, Prometheus Redis Exporter, redis-cli monitor для дебага.

Точка инициализации — один раз в lifespan FastAPI:

# adapters/out/cache/redis_client.py
from redis.asyncio import Redis
from app.config import Settings

_client: Redis | None = None


def get_redis(settings: Settings) -> Redis:
    return Redis.from_url(
        settings.redis.url,
        encoding="utf-8",
        decode_responses=True,
    )

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import Settings
from adapters.out.cache.redis_client import get_redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = Settings()
    app.state.redis = get_redis(settings)
    yield
    await app.state.redis.aclose()

app = FastAPI(lifespan=lifespan)

decode_responses=True — Redis возвращает str, а не bytes. Без этого каждый get придётся декодировать вручную.

JSON-сериализация — никогда pickle

R-CACHE-CFG-2: сериализация значений — JSON. pickle запрещён.

Почему не pickle:

  1. Security. pickle.loads() выполняет произвольный Python-код при десериализации. Если attacker может записать произвольный blob в Redis (через слабый key, через скомпрометированный соседний сервис) — RCE. Это тот же класс уязвимости, что Java JdkSerializationRedisSerializer.
  2. Читаемость. JSON в Redis viewable через redis-cli GET order-summaries:ORD-123 — видно содержимое для дебага. pickle-blob — нет.
  3. Совместимость. JSON переживает изменение Python-класса (добавили поле — старые значения читаются, новое поле None). pickle часто ломается при рефакторинге модели.

Пример через orjson (быстрее стандартного json):

# adapters/out/cache/json_serializer.py
import orjson
from typing import Any


def serialize(value: Any) -> str:
    return orjson.dumps(value).decode()


def deserialize(raw: str, model: type) -> Any:
    data = orjson.loads(raw)
    if hasattr(model, "model_validate"):
        return model.model_validate(data)
    return data

Для aiocache — явно указывать JsonSerializer:

from aiocache import Cache
from aiocache.serializers import JsonSerializer

cache = Cache(
    Cache.REDIS,
    endpoint="localhost",
    port=6379,
    serializer=JsonSerializer(),  # не PickleSerializer
)

Per-cache конфигурация

R-CACHE-CFG-3: каждый namespace — свой TTL.

Разные данные меняются с разной частотой:

NamespaceХарактер данныхTTL
customer-profilesредко меняется15 мин
product-catalogсправочник6 ч
feature-flagsfeature-store60 с
customer-balancesmoney15 с
order-summariesread-проекция5 мин

Один глобальный TTL = customer-balances stale 15 минут или product-catalog пересчитывается каждые 60 секунд. Оба варианта плохи. Подробнее — TTL.

pydantic-settings для cache settings

R-CACHE-CFG-4: настройки через pydantic-settings, не хардкод.

# app/config.py
from pydantic import BaseModel, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class CacheNamespaceSettings(BaseModel):
    ttl_seconds: int

    @field_validator("ttl_seconds")
    @classmethod
    def positive_ttl(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("ttl_seconds must be positive")
        return v


class RedisSettings(BaseModel):
    url: str = "redis://localhost:6379/0"


class CacheSettings(BaseModel):
    namespaces: dict[str, CacheNamespaceSettings]


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_nested_delimiter="__",
    )

    redis: RedisSettings = RedisSettings()
    cache: CacheSettings = CacheSettings(
        namespaces={
            "customer-profiles": CacheNamespaceSettings(ttl_seconds=900),
            "product-catalog": CacheNamespaceSettings(ttl_seconds=21600),
            "feature-flags": CacheNamespaceSettings(ttl_seconds=60),
            "customer-balances": CacheNamespaceSettings(ttl_seconds=15),
            "order-summaries": CacheNamespaceSettings(ttl_seconds=300),
        }
    )

В .env или переменных окружения:

REDIS__URL=redis://redis-master:6379/0
CACHE__NAMESPACES__CUSTOMER_BALANCES__TTL_SECONDS=10

SRE меняет TTL через config-map без redeploy. field_validator ловит невалидный ttl_seconds=0 при старте.

Кеш-порт как Protocol

В гексагональной архитектуре хендлер зависит от абстракции, не от redis.asyncio:

# core/ports/cache_port.py
from typing import Protocol, Any


class CachePort(Protocol):
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str, ttl_seconds: int) -> None: ...
    async def delete(self, key: str) -> None: ...

# adapters/out/cache/redis_cache_adapter.py
from redis.asyncio import Redis
from core.ports.cache_port import CachePort


class RedisCacheAdapter:
    def __init__(self, redis: Redis) -> None:
        self._redis = redis

    async def get(self, key: str) -> str | None:
        return await self._redis.get(key)

    async def set(self, key: str, value: str, ttl_seconds: int) -> None:
        await self._redis.set(key, value, ex=ttl_seconds)

    async def delete(self, key: str) -> None:
        await self._redis.delete(key)

Хендлер получает CachePort через DI — в тестах подменяется FakeCache, в проде — RedisCacheAdapter.

В тестах — fakeredis или Testcontainers

R-CACHE-CFG-5: два режима тестирования.

fakeredis — для unit/integration-тестов хендлеров, где нужно проверить «второй вызов не дёрнул репозиторий»:

# tests/conftest.py
import fakeredis.aioredis
import pytest_asyncio
from adapters.out.cache.redis_cache_adapter import RedisCacheAdapter


@pytest_asyncio.fixture
async def fake_cache():
    redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    yield RedisCacheAdapter(redis)
    await redis.aclose()

# tests/handlers/test_get_order_summary.py
async def test_second_call_uses_cache(fake_cache, order_repo):
    handler = GetOrderSummaryHandler(cache=fake_cache, repo=order_repo)

    await handler.execute(order_id="ORD-123")
    await handler.execute(order_id="ORD-123")

    order_repo.find_by_id.assert_called_once()  # второй вызов из кеша

Testcontainers — для тестов, проверяющих реальное TTL-поведение и eviction:

# tests/conftest.py
import pytest
from testcontainers.redis import RedisContainer
from redis.asyncio import Redis


@pytest.fixture(scope="session")
def redis_container():
    with RedisContainer("redis:7-alpine") as container:
        yield container


@pytest.fixture
async def real_redis(redis_container):
    client = Redis.from_url(
        redis_container.get_connection_url(),
        decode_responses=True,
    )
    yield client
    await client.flushdb()
    await client.aclose()

Мок CachePort через MagicMock теряет реальное поведение: TTL не истекает, eviction не происходит, тест проходит, в проде — баг.

aiocache-декораторы как альтернатива явному cache-aside

Для простых read-методов без сложной бизнес-логики aiocache сокращает шаблон:

from aiocache import cached
from aiocache.serializers import JsonSerializer
from app.config import Settings

settings = Settings()

@cached(
    ttl=settings.cache.namespaces["product-catalog"].ttl_seconds,
    key_builder=lambda f, *args, **kwargs: f"product-catalog:{kwargs['product_id']}",
    serializer=JsonSerializer(),
)
async def get_product_summary(product_id: str) -> dict:
    return await product_repo.find_summary(product_id)

Правила те же: JsonSerializer (не PickleSerializer), explicit key_builder (не автоматический из всех аргументов), TTL из настроек.

Что запрещено

АнтипаттернПравилоЧто взамен
pickle-сериализация (PickleSerializer)R-CACHE-CFG-X1JSON (JsonSerializer, orjson)
dict / lru_cache / cachetools в multi-instance продеR-CACHE-CFG-X2redis.asyncio
Один глобальный TTL для всех namespaceR-CACHE-CFG-X3per-namespace TTL в pydantic-settings
«Кеш» без реального backend (мок CachePort)R-CACHE-CFG-X4fakeredis или Testcontainers
TTL хардкодом в коде (ex=900)R-CACHE-CFG-4settings.cache.namespaces["..."].ttl_seconds
Redis.from_url(url) без decode_responses=True—decode_responses=True, иначе bytes
aiocache с PickleSerializerR-CACHE-CFG-X1явно serializer=JsonSerializer()

Куда дальше

  • Где кешируем — критерии выбора кандидатов, read-проекции vs агрегаты.
  • Ключи — namespace-префикс, explicit ключи, sensitive-данные.
  • TTL — типовые значения по характеру данных, money TTL.
  • Invalidation — evict на write, доменные события.
  • Паттерны — cache-aside, write-through, refresh-ahead.
  • Cache stampede — distributed lock, asyncio.Lock ограничения.
  • Observability — hit rate, prometheus-client, Redis Exporter.