Опирается на правила: R-CACHE-WHERE-1 … R-CACHE-WHERE-3 и R-CACHE-WHERE-X1 … R-CACHE-WHERE-X5 из Caching Rules → раздел 1. Где кешируем.

Важно знать

  • Кеш — оптимизация, не часть бизнес-контракта. Сервис обязан работать корректно и без кеша.
  • Если «кеш обязателен для корректности» — это бизнес-данные, их место в БД, не в кеше.
  • Read-heavy + редко меняющееся — кешируем: справочники (часы), user profile (~15 мин), feature flags (~60 сек), heavy aggregations.
  • Money — допустимо, но TTL ≤ 30 секунд + явный await cache.delete(key) на каждом write того же ресурса.
  • Cache-aside (lazy get-or-load + evict на write) — дефолтный паттерн в Python.
  • Доменный агрегат целиком — никогда: нарушает границы агрегата, invalidation становится неуправляемой. Кешируем read-проекции (OrderSummary).
  • Авторизация и валидация не кешируются руками: JWK кеш встроен (AUTH-5), ABAC-проверки каждый раз — мгновенный revoke.
  • В Python нет декларативного @Cacheable — cache-aside реализуется явно через cache-порт (Protocol в core/, реализация в adapters/out/cache/).

Кеш экономит latency и снижает нагрузку на БД, но добавляет stale-data риск и сложность invalidation. Главный вопрос перед добавлением кеша: «что произойдёт, если значение будет устаревшим на TTL, и кто это заметит?»

Read-heavy + редко меняющееся — кешируем

R-CACHE-WHERE-1: типичные кандидаты по характеру данных.

ЧтоTTLПример
Справочникичасыстраны, валюты, тарифы, типы документов
User profile / settings15–30 минутUserProfile, UserSettings
Feature flags30–60 секундFeatureFlagSet
Heavy aggregations5–10 минутDailyReport, TopProducts
JWK Set5 минутOAuth публичные ключи (AUTH-5 — встроен в библиотеку)

Критерий: ratio read/write > 100:1, данные не критичны к immediate consistency, бизнес-смысл допускает задержку на TTL.

В Python cache-aside реализуется через cache-порт. CachePort — Protocol в core/, конкретный Redis-адаптер — в adapters/out/cache/:

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

T = TypeVar("T")

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

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

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

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

    async def set(self, key: str, value: bytes, ttl: int) -> None:
        await self._client.set(key, value, ex=ttl)

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

Handler для GetUserProfile с явным cache-aside:

# application/use_cases/get_user_profile.py
import json
from dataclasses import dataclass
from core.ports.cache_port import CachePort
from core.ports.user_profile_repository import UserProfileRepository
from core.models.user_profile import UserProfile

@dataclass(frozen=True)
class GetUserProfileQuery:
    user_id: int

class GetUserProfileHandler:
    _CACHE_TTL = 900  # 15 минут

    def __init__(
        self,
        repo: UserProfileRepository,
        cache: CachePort,
    ) -> None:
        self._repo = repo
        self._cache = cache

    async def handle(self, query: GetUserProfileQuery) -> UserProfile:
        key = f"user-profiles:{query.user_id}"
        raw = await self._cache.get(key)
        if raw is not None:
            return UserProfile.model_validate_json(raw)

        profile = await self._repo.find_by_id(query.user_id)
        await self._cache.set(key, profile.model_dump_json().encode(), self._CACHE_TTL)
        return profile

Сериализация — JSON через model_dump_json() / model_validate_json(). pickle — запрещён (R-CACHE-CFG-X1).

Money — допустимо, но строго

R-CACHE-WHERE-2: баланс, лимиты, available credit можно кешировать только с явной invalidation strategy.

  • TTL 5–30 секунд (не больше).
  • await cache.delete(key) на каждом write-методе того же ресурса.
  • Для критичных операций (списание, перевод) — evict перед чтением (защита от race между двумя процессами).
# application/use_cases/get_customer_balance.py
@dataclass(frozen=True)
class GetCustomerBalanceQuery:
    customer_id: int

class GetCustomerBalanceHandler:
    _CACHE_TTL = 15  # 15 секунд

    def __init__(self, repo: CustomerBalanceRepository, cache: CachePort) -> None:
        self._repo = repo
        self._cache = cache

    def _key(self, customer_id: int) -> str:
        return f"customer-balances:{customer_id}"

    async def handle(self, query: GetCustomerBalanceQuery) -> CustomerBalance:
        key = self._key(query.customer_id)
        raw = await self._cache.get(key)
        if raw is not None:
            return CustomerBalance.model_validate_json(raw)

        balance = await self._repo.find_by_customer(query.customer_id)
        await self._cache.set(key, balance.model_dump_json().encode(), self._CACHE_TTL)
        return balance

# application/use_cases/debit_customer_balance.py
@dataclass(frozen=True)
class DebitCustomerBalanceCommand:
    customer_id: int
    amount: Decimal

class DebitCustomerBalanceHandler:
    def __init__(self, repo: CustomerBalanceRepository, cache: CachePort) -> None:
        self._repo = repo
        self._cache = cache

    def _key(self, customer_id: int) -> str:
        return f"customer-balances:{customer_id}"

    async def handle(self, command: DebitCustomerBalanceCommand) -> CustomerBalance:
        key = self._key(command.customer_id)
        await self._cache.delete(key)          # evict before read — защита от race

        balance = await self._repo.find_by_customer(command.customer_id)
        updated = balance.subtract(command.amount)
        await self._repo.save(updated)

        await self._cache.delete(key)          # evict after write
        return updated

Double-evict (до и после транзакции) защищает от race: другой запрос мог положить старое значение в кеш между первым evict и save.

Cache-aside — дефолтный паттерн

R-CACHE-WHERE-3: read проходит через cache-read, write делает явный evict.

Request → cache.get(key)
              │
              ├── HIT  → return cached
              │
              └── MISS → load from DB → cache.set(key, value, ttl) → return

Write → DB update → cache.delete(key) → следующий read загружает свежее

Подробнее о вариантах реализации — Паттерны.

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

Кеш на write-методах

R-CACHE-WHERE-X1: кешировать результат create_order(), confirm_payment() — нонсенс.

# КАТАСТРОФА
_cache: dict = {}

async def create_order(command: CreateOrderCommand) -> Order:
    key = f"orders:{command.customer_id}"
    if key in _cache:
        return _cache[key]          # возвращает Order из прошлого вызова
    order = await order_repo.save(Order.from_command(command))
    _cache[key] = order
    return order

Сценарий: клиент шлёт CreateOrderCommand(customer_id=42, ...). Первый вызов проходит, создаёт заказ. Второй вызов с тем же customer_id — возвращает закешированный Order из первого. Реально новый заказ не создаётся, но клиент думает, что создан.

Cache-aside имеет смысл только для read: «для тех же параметров — тот же результат». Write порождает side-effects, кеш их прячет.

Кеш доменного агрегата целиком

R-CACHE-WHERE-X2: кешировать Order целиком с items, payment, shipment.

Проблемы:

  • Нарушает границы агрегата. Агрегат управляет собственными invariants; кеш видит снимок, который уже может быть нарушен дочерними изменениями.
  • Sensitive data. Агрегат может содержать поля, не предназначенные для кеша (PAN, PII).
  • Invalidation hell. Любое изменение OrderItem, Payment, Shipment должно evict-ить родительский Order. Легко пропустить.

Корректно — read-проекции:

# core/models/order_summary.py
from pydantic import BaseModel
from decimal import Decimal
from enum import StrEnum

class OrderStatus(StrEnum):
    PENDING = "PENDING"
    CONFIRMED = "CONFIRMED"
    SHIPPED = "SHIPPED"

class OrderSummary(BaseModel):
    order_id: int
    customer_id: int
    customer_name: str
    status: OrderStatus
    item_count: int
    total_amount: Decimal

# application/use_cases/get_order_summary.py
@dataclass(frozen=True)
class GetOrderSummaryQuery:
    order_id: int

class GetOrderSummaryHandler:
    _CACHE_TTL = 300  # 5 минут

    def __init__(self, repo: OrderSummaryRepository, cache: CachePort) -> None:
        self._repo = repo
        self._cache = cache

    async def handle(self, query: GetOrderSummaryQuery) -> OrderSummary:
        key = f"order-summaries:{query.order_id}"
        raw = await self._cache.get(key)
        if raw is not None:
            return OrderSummary.model_validate_json(raw)

        summary = await self._repo.find_by_id(query.order_id)
        await self._cache.set(key, summary.model_dump_json().encode(), self._CACHE_TTL)
        return summary

OrderSummary — компактная Pydantic-модель, без sensitive-полей, без вложенных коллекций. Кешируется предсказуемо.

Money без TTL/invalidation

R-CACHE-WHERE-X3: «кешируем баланс пользователя Сбера на час».

Сценарий: пользователь потратил, баланс в БД 0 ₽, в кеше 50 000 ₽ ещё 55 минут. Открывает приложение → видит 50 000 ₽ → пытается потратить → отказ → обращение в поддержку.

Money — отдельный класс данных:

  • TTL ≤ 30 секунд.
  • Явный evict на каждой write-операции.
  • Double-evict для критичных flows (до и после транзакции).

Кеш бизнес-критичных данных без trade-off

R-CACHE-WHERE-X4: «кешируем статус заказа, потому что быстрее» — без оценки stale-data риска.

Перед добавлением кеша три вопроса:

  1. Что произойдёт, если значение устареет на TTL? Кто заметит?
  2. Бизнес это терпит? (Список валют — да; статус Product.available перед списанием — нет.)
  3. Есть ли явная invalidation strategy?

Три «да» — кешируем. Иначе — нет.

Кеш авторизации/валидации

R-CACHE-WHERE-X5: кешировать «у пользователя customer_id=42 есть роль MANAGER» на 10 минут — security risk.

Сценарий: в 10:00 роль отозвали. До 10:10 пользователь продолжает выполнять действия, доступные MANAGER, потому что кеш.

JWK кеш встроен в OIDC-библиотеку. ABAC-проверки (проверка прав в хендлере) делаются каждый раз — цена несколько мс, выгода — мгновенный revoke.

Что запрещено — таблица

АнтипаттернПравилоЧто взамен
Кеш на write-методе (create_order, confirm_payment)R-CACHE-WHERE-X1только read; write — await cache.delete(key)
Кеш доменного агрегата целикомR-CACHE-WHERE-X2read-проекции (OrderSummary)
Money-кеш без TTL/invalidationR-CACHE-WHERE-X3TTL ≤ 30 сек + evict before/after write
Кеш бизнес-данных без оценки trade-offR-CACHE-WHERE-X4три вопроса перед cache-aside
Кеш JWT/ABAC-валидации рукамиR-CACHE-WHERE-X5JWK встроен; ABAC — каждый раз
pickle вместо JSONR-CACHE-CFG-X1model_dump_json() / model_validate_json()
In-memory dict в multi-instance продеR-CACHE-CFG-X2Redis (redis.asyncio)

Куда дальше

  • Конфигурация — redis.asyncio, JSON-сериализация, per-cache TTL через pydantic-settings.
  • TTL — типовые значения TTL, политики истечения.
  • Invalidation — явный evict, invalidation как side-effect доменного события.
  • Паттерны — cache-aside vs write-through vs refresh-ahead.
  • Cache stampede — distributed lock через redis.lock, refresh-ahead.
  • Ключи — namespace-префикс, explicit key, хеширование sensitive.
  • Observability — hit rate метрика, prometheus-client.