Ключи кеша — namespace, f-string и хеширование чувствительных данных

Ключи кеша в Python: namespace-префикс через имя кеша, explicit f-string вместо repr(), kebab-case, composite через ':', SHA-256 для email и телефонов.

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

Опирается на правила: R-CACHE-KEY-1 … R-CACHE-KEY-4 и R-CACHE-KEY-X1 … R-CACHE-KEY-X4 из Caching Rules → раздел 3. Ключи.

Важно знать

Namespace через имя кеша — сборка вручную: f"user-profiles:{user_id}". Автоматического префикса, как в Spring, нет.

Имена в kebab-case: user-profiles, payment-methods, feature-flags. Не snake_case, не camelCase.

Ключ всегда explicit — f-string с конкретными полями. Никогда repr(obj) или id(obj).

Composite ключ — поля через : (Redis-конвенция вложенных namespaces).

Кастомная сборка ключа выносится в отдельную функцию — только для 4+ полей или нестандартной нормализации.

Sensitive данные (email, телефон, токен) в ключе — только через SHA-256, не plain.

Один cache — один тип данных. Общий cache shared ломает TTL, метрики и invalidation.

Ключ — то, по чему cache находит значение. В Python нет декларативного @Cacheable с автогенерацией ключа — вся сборка происходит явно в коде. Это одновременно риск (легко допустить ошибку) и контроль (всегда понятно, что именно попадёт в Redis).

Namespace через префикс имени кеша

R-CACHE-KEY-1: namespace собирается вручную как часть ключа.

CACHE_USER_PROFILES = "user-profiles"

async def find_profile(user_id: int, cache: CachePort) -> UserProfile | None:
    key = f"{CACHE_USER_PROFILES}:{user_id}"
    cached = await cache.get(key)
    if cached is not None:
        return UserProfile.model_validate_json(cached)
    profile = await db.fetch_user_profile(user_id)
    if profile:
        await cache.set(key, profile.model_dump_json(), ttl=settings.cache.user_profiles_ttl)
    return profile

В Redis ключ: user-profiles:42. Константа CACHE_USER_PROFILES — единственное место, где хранится имя namespace. Любой evict использует её же:

await cache.delete(f"{CACHE_USER_PROFILES}:{user_id}")

Что даёт namespace:

KEYS user-profiles:* — все ключи этого кеша одним scan.
Evict-by-namespace без затрагивания других кешей.
Метрики per-cache, если инструментируется на уровне первого сегмента ключа.

Kebab-case slug для имени

R-CACHE-KEY-2: имена в kebab-case.

user-profiles       ✓
payment-methods     ✓
feature-flags       ✓
order-summaries     ✓
customer-balances   ✓
user_profiles       ✗ — snake_case
userProfiles        ✗ — camelCase
UserProfiles        ✗ — PascalCase

Соглашение: Redis-ключи в UCP идут в slug-формате, как URL paths. Это даёт согласованность между именами кешей, путями API (/user-profiles/42) и Kafka-топиками (user-profile.updated).

Explicit f-string key

R-CACHE-KEY-3: ключ — f-string с явными полями, не repr() и не целый объект.

Один параметр:

CACHE_ORDERS = "order-summaries"

async def get_order_summary(order_id: int, cache: CachePort) -> OrderSummary | None:
    key = f"{CACHE_ORDERS}:{order_id}"
    ...

Composite ключ — поля через ::

CACHE_CUSTOMER_ORDERS = "orders-by-customer"

async def find_by_customer_and_status(
    customer_id: int,
    status: OrderStatus,
    cache: CachePort,
) -> list[OrderSummary]:
    key = f"{CACHE_CUSTOMER_ORDERS}:{customer_id}:{status.value}"
    ...

Redis key: orders-by-customer:42:CONFIRMED. Разделитель : — Redis-конвенция вложенных namespaces.

Объект-параметр — только явные поля, не repr(query):

CACHE_PRODUCT_SEARCH = "product-search"

async def search_products(query: ProductSearchQuery, cache: CachePort) -> list[ProductSummary]:
    key = f"{CACHE_PRODUCT_SEARCH}:{query.category}:{query.min_price}:{query.max_price}:{query.page}"
    cached = await cache.get(key)
    if cached is not None:
        return [ProductSummary.model_validate(item) for item in json.loads(cached)]
    ...

Никогда key = repr(query) или key = str(query) — repr() для Pydantic-модели включает адрес объекта или нестабильный порядок полей, hit rate стремится к нулю.

Кастомная сборка ключа

R-CACHE-KEY-4: для 4+ полей или нестандартной нормализации выносим логику в отдельную функцию.

from dataclasses import dataclass
from datetime import date

CACHE_ORDER_SEARCH = "order-search"

@dataclass(frozen=True)
class OrderSearchCriteria:
    customer_id: int
    status: OrderStatus
    from_date: date
    to_date: date
    page: int
    size: int

def build_order_search_key(criteria: OrderSearchCriteria) -> str:
    return ":".join([
        CACHE_ORDER_SEARCH,
        str(criteria.customer_id),
        criteria.status.value,
        criteria.from_date.isoformat(),
        criteria.to_date.isoformat(),
        str(criteria.page),
        str(criteria.size),
    ])

Использование:

async def search_orders(criteria: OrderSearchCriteria, cache: CachePort) -> list[OrderSummary]:
    key = build_order_search_key(criteria)
    cached = await cache.get(key)
    ...

Когда оправдан: критериев 4 и больше, нужна нормализация (lowercase, truncation дат, сортировка списков), нужна тестируемость логики ключа отдельно от cache-логики.

build_order_search_key легко покрывается unit-тестом без Redis.

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

Автогенерация ключа из всех аргументов

R-CACHE-KEY-X1: aiocache может сгенерировать ключ из имени функции и аргументов — по умолчанию ненадёжно.

# ОПАСНО — ключ включает repr всех аргументов
@cached(ttl=900, namespace="orders")
async def find_by_customer_and_status(customer_id: int, status: OrderStatus) -> list[OrderSummary]:
    ...

Что ломается:

Переименовали функцию — ключ изменился, кеш стал мёртвым.
Изменили порядок параметров — hit rate упал.
status — enum, repr() может включать имя класса (<OrderStatus.CONFIRMED: 'CONFIRMED'>).

Используй key_builder явно:

def customer_orders_key(func, customer_id: int, status: OrderStatus) -> str:
    return f"orders-by-customer:{customer_id}:{status.value}"

@cached(ttl=900, key_builder=customer_orders_key)
async def find_by_customer_and_status(customer_id: int, status: OrderStatus) -> list[OrderSummary]:
    ...

`repr()` или `id()` объекта в ключе

R-CACHE-KEY-X2: для Pydantic-моделей repr() содержит нестабильные части.

CACHE_PRODUCT_SEARCH = "product-search"

# КАТАСТРОФА — каждый вызов = новый ключ
async def search_products(query: ProductSearchQuery, cache: CachePort) -> list[ProductSummary]:
    key = f"{CACHE_PRODUCT_SEARCH}:{repr(query)}"
    ...

repr() Pydantic v2 для модели ProductSearchQuery(category='electronics', page=1) выглядит как ProductSearchQuery(category='electronics', page=1) — стабильно только при неизменном порядке полей в модели. Добавили поле с дефолтом — repr изменился, кеш мёртв. Используй явные поля, как показано в разделе выше.

Один общий cache `shared`

R-CACHE-KEY-X3: соблазн положить всё в один namespace.

CACHE_SHARED = "shared"

# ПЛОХО — OrderSummary и UserProfile в одном кеше
async def get_order(order_id: int, cache: CachePort) -> OrderSummary | None:
    key = f"{CACHE_SHARED}:order:{order_id}"
    ...

async def get_profile(user_id: int, cache: CachePort) -> UserProfile | None:
    key = f"{CACHE_SHARED}:user:{user_id}"
    ...

Проблемы:

TTL общий — профиль (15 минут) и заказ (60 секунд) не могут жить в одном кеше с разными TTL если используется namespace-уровневый TTL.
Метрики смешанные — hit/miss по кешу нечитаемы.
Evict-by-namespace удаляет всё сразу.

Per-entity:

CACHE_ORDER_SUMMARIES = "order-summaries"
CACHE_USER_PROFILES = "user-profiles"

async def get_order(order_id: int, cache: CachePort) -> OrderSummary | None:
    key = f"{CACHE_ORDER_SUMMARIES}:{order_id}"
    ...

async def get_profile(user_id: int, cache: CachePort) -> UserProfile | None:
    key = f"{CACHE_USER_PROFILES}:{user_id}"
    ...

Plain PII в ключе

R-CACHE-KEY-X4: Redis-ключи видны в MONITOR, slow query log, snapshot, network capture при non-TLS.

CACHE_CUSTOMER_BY_EMAIL = "customer-by-email"

# ПЛОХО — email plain в ключе
async def find_customer_by_email(email: str, cache: CachePort) -> CustomerProfile | None:
    key = f"{CACHE_CUSTOMER_BY_EMAIL}:{email}"
    ...

# ХОРОШО — SHA-256
import hashlib

async def find_customer_by_email(email: str, cache: CachePort) -> CustomerProfile | None:
    email_hash = hashlib.sha256(email.lower().encode()).hexdigest()[:16]
    key = f"{CACHE_CUSTOMER_BY_EMAIL}:{email_hash}"
    ...

Лучшее решение — вообще не использовать email как cache-ключ. Внутренний customer_id (integer) идеален: не несёт PII, уникален, компактен. Поиск по email → сначала получить customer_id, потом — f"customer-profiles:{customer_id}".

Аналогично для номера телефона в Sber-подобных сценариях:

CACHE_SBER_CLIENT = "sber-client-profile"

async def get_client_profile(phone: str, cache: CachePort) -> SberClientProfile | None:
    phone_hash = hashlib.sha256(phone.encode()).hexdigest()[:16]
    key = f"{CACHE_SBER_CLIENT}:{phone_hash}"
    ...

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

Антипаттерн	Правило	Что взамен
Автогенерация ключа без `key_builder`	`R-CACHE-KEY-X1`	explicit `key_builder` или f-string
`repr(obj)` / `str(obj)` в ключе	`R-CACHE-KEY-X2`	f-string с явными полями модели
Один общий cache `shared` для разных entity	`R-CACHE-KEY-X3`	per-entity namespace-константа
Email / телефон / токен plain в ключе	`R-CACHE-KEY-X4`	SHA-256 или внутренний integer ID
snake_case / camelCase имя кеша	`R-CACHE-KEY-2`	kebab-case slug
`repr(query)` для Pydantic-модели	`R-CACHE-KEY-X2`	`f"...:{query.field_a}:{query.field_b}"`
Нет namespace-константы, строка в коде	`R-CACHE-KEY-1`	`CACHE_NAME = "..."` в одном месте
Разделитель `_` в composite key	`R-CACHE-KEY-3`	`:` (Redis-конвенция)

Куда дальше

Cache stampede — distributed lock в asyncio при общем miss.
Конфигурация — per-cache TTL через pydantic-settings.
Invalidation — evict по тем же ключам, что и cache-aside.
Observability — метрики hit/miss через prometheus-client.
Паттерны — cache-aside, write-through, refresh-ahead.
TTL — типовые значения TTL по характеру данных.
Где кешируем — какие данные ключевать, а какие нет.