TTL — explicit duration per cache и типовые значения по характеру данных (Python)

TTL в Python-кешировании через redis.asyncio и aiocache: каждый кеш получает explicit TTL из pydantic-settings, никаких infinite-кешей и pickle-сериализации.

Опубликовано 19 июня 2026 ~5 мин чтения Автор: Вадим Викулин

Опирается на правила: R-CACHE-TTL-1 … R-CACHE-TTL-4 и R-CACHE-TTL-X1 … R-CACHE-TTL-X3 из Caching Style Guide → раздел 4. TTL.

Важно знать

Каждый кеш — explicit TTL. Никаких infinite-кешей и None как «навсегда».

Типовые значения по характеру: static reference — часы; user profile — 15 мин; feature flags — 60 сек; heavy aggregations — 5-10 мин; money — 5-30 сек.

TTL — через pydantic-settings, не хардкод в коде. Ops меняет под нагрузку без redeploy.

Если есть естественный invalidation event — TTL длиннее; invalidation делает основную работу.

None / 0 = infinite в redis.asyncio и aiocache — Redis при max-memory eviction по LRU без вашего контроля.

TTL > 24 часов — кеш переживает деплой с устаревшей структурой Pydantic-модели → ValidationError на проде.

Money без TTL — главный антипаттерн: списали, баланс в кеше старый → инцидент.

Pickle — запрещён: R-CACHE-CFG-X1, RCE при десериализации недоверенных данных; всегда JSON.

TTL — единственная защита, когда invalidation не реализован или пропущен. Если invalidation надёжный, TTL — резерв; если нет — TTL единственное, что не даёт кешу жить вечно.

Explicit TTL на каждом кеше

R-CACHE-TTL-1: ноль infinite-кешей в проде.

В Python нет декларативного @Cacheable с per-cache TTL из конфига, как в Spring. Паттерн — кеш-порт (Protocol в core/) + реализация через redis.asyncio в adapters/out/cache/, TTL передаётся при записи явно.

Настройки через pydantic-settings (R-CACHE-CFG-4):

from datetime import timedelta
from pydantic import Field
from pydantic_settings import BaseSettings


class CacheSettings(BaseSettings):
    redis_url: str = Field(default="redis://localhost:6379/0")
    ttl_user_profiles: timedelta = Field(default=timedelta(minutes=15))
    ttl_currencies: timedelta = Field(default=timedelta(hours=6))
    ttl_feature_flags: timedelta = Field(default=timedelta(seconds=60))
    ttl_user_balances: timedelta = Field(default=timedelta(seconds=30))
    ttl_order_summaries: timedelta = Field(default=timedelta(minutes=5))
    ttl_top_products: timedelta = Field(default=timedelta(minutes=10))

    model_config = {"env_prefix": "CACHE_"}

Порт кеша в core/:

from typing import Protocol, Any


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

Реализация через redis.asyncio в adapters/out/cache/:

import json
from datetime import timedelta
from typing import Any

import redis.asyncio as aioredis

from core.ports.cache_port import CachePort


class RedisCache(CachePort):
    def __init__(self, client: aioredis.Redis) -> None:
        self._client = client

    async def get(self, key: str) -> Any | None:
        raw = await self._client.get(key)
        if raw is None:
            return None
        return json.loads(raw)

    async def set(self, key: str, value: Any, ttl: timedelta) -> None:
        await self._client.set(key, json.dumps(value), ex=int(ttl.total_seconds()))

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

ex=int(ttl.total_seconds()) — целые секунды, никаких None.

Типовые значения по характеру данных

R-CACHE-TTL-2: ориентируемся на природу данных, не на «короче — лучше».

Тип данных	TTL	Пример
Static reference	часы	currencies, countries, timezones
User profile / preferences	15 мин	`CustomerProfile`, `UserSettings`
Feature flags	30-60 сек	`FeatureFlagSet`, A/B test buckets
Configuration overrides	60 сек	`TenantConfig`, runtime settings
Heavy aggregations	5-10 мин	`DailyReport`, `TopProducts`
Money-related	5-30 сек	`CustomerBalance` (только с явным evict)

Логика:

Short TTL (секунды) — данные часто меняются, stale ощутим.
Medium TTL (минуты) — изменения редкие; задержка 15 минут терпима.
Long TTL (часы) — справочники, практически неизменные; редкие изменения требуют ручного delete.

Пример cache-aside для CustomerProfile с TTL из настроек:

from dataclasses import dataclass
from datetime import timedelta

from core.ports.cache_port import CachePort
from core.domain.customer import CustomerProfile
from config.cache_settings import CacheSettings


@dataclass
class GetCustomerProfileHandler:
    cache: CachePort
    customer_repo: CustomerRepository
    settings: CacheSettings

    async def handle(self, customer_id: int) -> CustomerProfile:
        key = f"customer-profiles:{customer_id}"
        cached = await self.cache.get(key)
        if cached is not None:
            return CustomerProfile.model_validate(cached)

        profile = await self.customer_repo.find_by_id(customer_id)
        await self.cache.set(
            key,
            profile.model_dump(),
            ttl=self.settings.ttl_user_profiles,
        )
        return profile

TTL приходит из settings.ttl_user_profiles — изменение через переменную окружения CACHE_TTL_USER_PROFILES=5m без redeploy.

TTL в настройках, не в коде

R-CACHE-TTL-3: timedelta(minutes=15) зашитый прямо в хендлере — антипаттерн.

# ПЛОХО — ops не может поменять без redeploy
await self.cache.set(key, data, ttl=timedelta(minutes=15))

Когда SRE фиксирует инциденты из-за stale-профиля — нужен новый деплой, чтобы сменить 15m на 5m. Это десятки минут (CI + rollout).

# ХОРОШО — переменная окружения в ConfigMap/Secrets
CACHE_TTL_USER_PROFILES=5m

Pydantic автоматически парсит timedelta из строк вида "5m", "30s", "6h" при использовании плагина pydantic[extra] или ручной валидации. Альтернатива — хранить TTL в секундах как int и оборачивать в timedelta:

class CacheSettings(BaseSettings):
    ttl_user_profiles_sec: int = Field(default=900)

    @property
    def ttl_user_profiles(self) -> timedelta:
        return timedelta(seconds=self.ttl_user_profiles_sec)

Оба подхода работают; главное — TTL не в коде хендлера.

TTL и invalidation — два слоя защиты

R-CACHE-TTL-4: если у данных есть естественный invalidation event — TTL может быть длиннее.

Сценарий 1 — есть событие (OrderConfirmed):

@dataclass
class OrderConfirmedEventHandler:
    cache: CachePort

    async def handle(self, event: OrderConfirmedEvent) -> None:
        await self.cache.delete(f"order-summaries:{event.order_id}")

TTL для order-summaries — 30 минут. Invalidation срабатывает быстро; TTL — резерв на случай пропущенного события.

Сценарий 2 — события нет (внешний справочник, изменения вне сервиса):

# Курсы валют меняются через ETL ночью — нет внутреннего события
# TTL = 6ч, stale до 6 часов терпимо для отображения курса
key = f"currencies:{currency_code}"
await self.cache.set(key, rate_data, ttl=self.settings.ttl_currencies)

Принцип: invalidation первичен, TTL вторичен. Если invalidation надёжен — TTL длиннее. Нет invalidation — TTL короче.

Money-данные: короткий TTL обязателен

R-CACHE-TTL-X3: CustomerBalance с TTL > 1 минуты без строгой invalidation = инцидент.

@dataclass
class GetCustomerBalanceHandler:
    cache: CachePort
    balance_repo: BalanceRepository
    settings: CacheSettings

    async def handle(self, customer_id: int) -> CustomerBalance:
        key = f"customer-balances:{customer_id}"
        cached = await self.cache.get(key)
        if cached is not None:
            return CustomerBalance.model_validate(cached)

        balance = await self.balance_repo.find_by_customer(customer_id)
        await self.cache.set(
            key,
            balance.model_dump(),
            ttl=self.settings.ttl_user_balances,  # 30 секунд
        )
        return balance

Evict на write-операции:

@dataclass
class ProcessPaymentHandler:
    cache: CachePort
    payment_service: PaymentService

    async def handle(self, command: ProcessPaymentCommand) -> None:
        await self.payment_service.process(command)
        await self.cache.delete(f"customer-balances:{command.customer_id}")

Для критичных flows — delete перед чтением, не только после write:

async def handle(self, customer_id: int) -> CustomerBalance:
    key = f"customer-balances:{customer_id}"
    await self.cache.delete(key)  # evict перед критичным чтением
    balance = await self.balance_repo.find_by_customer(customer_id)
    await self.cache.set(key, balance.model_dump(), ttl=self.settings.ttl_user_balances)
    return balance

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

Антипаттерн	Правило	Что взамен
`ex=None` / `ex=0` (infinite) в `redis.asyncio`	`R-CACHE-TTL-X1`	explicit `ex=int(ttl.total_seconds())`
TTL > 24 часов для бизнес-данных	`R-CACHE-TTL-X2`	TTL ≤ 24ч или version-suffix в ключе
Money без TTL	`R-CACHE-TTL-X3`	TTL ≤ 30с + явный `delete` на write
`timedelta(...)` хардкодом в хендлере	`R-CACHE-TTL-3`	`pydantic-settings` + переменная окружения
Один TTL на все кеши	`R-CACHE-TTL-1`	per-cache поле в `CacheSettings`
TTL без учёта характера данных	`R-CACHE-TTL-2`	таблица типовых значений
Long TTL без invalidation event	`R-CACHE-TTL-4`	short TTL либо надёжный evict-handler
`pickle` как сериализатор	`R-CACHE-CFG-X1`	`json.dumps` / `json.loads`

Куда дальше

Cache stampede — asyncio.Lock и redis.lock при истечении TTL под нагрузкой.
Конфигурация — сборка RedisCache-адаптера и CacheSettings.
Invalidation — delete на write, event-driven evict.
Ключи — namespace-префиксы, kebab-case, sensitive-данные в ключах.
Observability — hit rate через prometheus-client, мониторинг реального TTL.
Паттерны — refresh-ahead для hot-ключей (TTL менее критичен).
Где кешируем — money-rules, read-проекции, не агрегаты.