← назад к разделу

Redis в Python-сервисе появляется по трём поводам: кэш перед PostgreSQL, короткоживущее состояние (сессии, лимиты, блокировки) и очереди или потоки между процессами. Клиент для всех трёх один, redis-py с модулем redis.asyncio, и в отличие от фреймворков с аннотациями он не прячет ни одной команды: кэш, инвалидация, таймауты и поведение при падении Redis это код, который пишете вы. Это и есть предмет статьи: не команды Redis (они разобраны в соседних статьях раздела), а то, как их правильно обернуть в сервисе.

Обязательно

Клиент и таймауты

from redis.asyncio import Redis
from redis.asyncio.retry import Retry
from redis.backoff import ExponentialBackoff
from redis.exceptions import ConnectionError, TimeoutError


def new_client(url: str) -> Redis:
    return Redis.from_url(
        url,
        socket_connect_timeout=1,
        socket_timeout=0.3,
        max_connections=20,
        health_check_interval=30,
        retry=Retry(ExponentialBackoff(cap=0.1, base=0.01), retries=1),
        retry_on_error=[ConnectionError, TimeoutError],
        client_name="orders",
    )

Один Redis на процесс: внутри пул соединений, он безопасен для одновременных корутин; открывают его в lifespan, закрывают через await redis.aclose(). Значения выше отличаются от умолчаний не случайно.

  • Таймауты на сокете по умолчанию 5 секунд и на соединение, и на ответ. Для кэша это приговор: Redis завис, и каждый запрос к сервису ждёт пять секунд, прежде чем пойти в базу. Сотни миллисекунд это уже много для хранилища, которое отвечает за доли миллисекунды; таймаут кэша должен быть меньше, чем ожидание запроса в базу.
  • Повторы по умолчанию — десять с нарастающими паузами до секунды. Недоступный Redis с такими настройками даёт ошибку только через несколько секунд на каждую команду. На пути запроса пользователя повтор это удвоенная задержка при проблеме, поэтому один или ноль; десять оставляют фоновым задачам. Список retry_on_error говорит, что именно повторять: без TimeoutError в нём таймауты не повторяются вовсе.
  • Размер пула по умолчанию 100 соединений от каждого экземпляра сервиса. Redis однопоточный, ему от этого не легче; 10–20 соединений хватает почти всем. Когда свободных нет, обычный пул сразу бросает MaxConnectionsError, а BlockingConnectionPool ждёт освобождения с таймаутом — второй честнее под всплеском.
  • Дедлайн запроса не ограничивает команду сам по себе. Обработчик FastAPI может быть отменён, но команда в сокете продолжит ждать свой socket_timeout. Отмена задачи посреди команды безопасна: клиент закроет это соединение, а не оставит его с недочитанным ответом, но времени она не экономит, поэтому таймауты задают на клиенте.

from_url разбирает строку подключения вида redis://:secret@redis:6379/0, и это удобно для конфигурации через окружение; клиент говорит по RESP2, protocol=3 включает RESP3 с типизированными ответами. Для Sentinel и Cluster есть свои классы, о них ниже; код сервиса от них зависит только через общий набор команд.

Что хранить и как сериализовать

Redis хранит байты. Модель сервиса превращают в них явно: JSON через model_dump_json() для читаемости и совместимости, msgpack, когда счёт идёт на мегабайты в секунду. Одно значение целиком удобнее хеша, пока его читают и пишут целиком; хеш (HSET/HGETALL) выигрывает, когда нужно обновлять одно поле или читать часть. Ключ строят по схеме сущность:идентификатор, и схему держат в одном месте кода, а не размазывают по вызовам.

class CachedProduct(BaseModel):
    id: int
    name: str
    price: int

Деньги в копейках целым числом, время в Unix-секундах или ISO 8601: JSON от Python и JSON от сервиса на другом языке должны читать друг друга, если кэш общий. decode_responses оставляют выключенным и разбирают байты сами: так одна и та же модель читает и то, что положил этот сервис, и то, что положил сосед.

Cache-aside: чтение сквозь кэш

Самый частый узор: посмотреть в Redis, при промахе сходить в базу и положить результат с TTL.

GET order:42 попадание вернуть из Redis промах прочитать из базы, SET с TTL изменили заказ в базе DEL order:42

Чтение сначала спрашивает кэш и при промахе кладёт результат с временем жизни; запись инвалидирует ключ удалением, а не обновлением.

import asyncio
import random
from collections.abc import Awaitable, Callable

from pydantic import ValidationError


class ProductCache:
    def __init__(self, redis: Redis, load: Callable[[int], Awaitable[CachedProduct]], ttl: int = 300) -> None:
        self.redis = redis
        self.load = load
        self.ttl = ttl
        self.inflight: dict[str, asyncio.Future[CachedProduct]] = {}

    async def get(self, product_id: int) -> CachedProduct:
        key = f"product:{product_id}"
        try:
            raw = await self.redis.get(key)
            if raw is not None:
                return CachedProduct.model_validate_json(raw)
        except (ConnectionError, TimeoutError, ValidationError) as e:
            log.warning("redis get failed", key=key, error=type(e).__name__)
        return await self._load_once(key, product_id)

    async def _load_once(self, key: str, product_id: int) -> CachedProduct:
        if key in self.inflight:
            return await self.inflight[key]
        future: asyncio.Future[CachedProduct] = asyncio.get_running_loop().create_future()
        self.inflight[key] = future
        try:
            product = await self.load(product_id)
            future.set_result(product)
        except BaseException as e:
            future.set_exception(e)
            raise
        finally:
            del self.inflight[key]
        try:
            await self.redis.set(key, product.model_dump_json(), ex=self.ttl + random.randint(0, 30))
        except (ConnectionError, TimeoutError) as e:
            log.warning("redis set failed", key=key, error=type(e).__name__)
        return product

Здесь четыре решения, и каждое закрывает отдельную аварию.

  • None это промах, а не ошибка. get отсутствующего ключа возвращает None; любая ошибка клиента (таймаут, обрыв) логируется, и запрос идёт в базу так же, как при промахе. Кэш, который при собственной ошибке роняет запрос, хуже отсутствия кэша.
  • Один запрос в базу на ключ. Когда ключ истёк, а товар популярный, сто корутин одновременно идут в базу за одним и тем же. Словарь inflight с Future на ключ пускает в базу первую, остальные ждут её результат. Это защита от лавины на уровне процесса; между процессами её дополняет короткий TTL и база, которая выдерживает пару десятков одинаковых запросов. Готовой библиотеки на эту роль, как singleflight в других стеках, в Python нет, и двадцать строк выше — вся реализация.
  • TTL с разбросом. Если тысяча ключей положена в одну секунду с одинаковым TTL, через час они истекут в одну секунду. Небольшой разброс размазывает этот всплеск.
  • Испорченное значение равно промаху. Не разобралась модель (ValidationError после смены структуры на выкате) значит перечитать из базы и перезаписать, а не вернуть 500.

Отсутствие товара тоже стоит кэшировать: короткий ключ-метка на 30–60 секунд, иначе запросы по несуществующим идентификаторам (боты, старые ссылки) всегда долетают до базы.

Инвалидация: удалять, а не обновлять

При изменении товара кэш не обновляют новым значением, а удаляют ключ: следующий читатель положит свежее. Обновление на запись проигрывает в гонке «два обработчика изменили товар, в кэш последним записал тот, кто читал базу раньше». Удаление делают после коммита транзакции в базе, а не до: иначе параллельный читатель успеет положить в кэш старое значение между удалением и коммитом.

async def invalidate(self, product_id: int) -> None:
    await self.redis.delete(f"product:{product_id}")

Гонка остаётся и здесь (читатель прочитал из базы старое, в это время прошли коммит и удаление, читатель записал старое в кэш), её закрывает только короткий TTL: на пять минут старая цена допустима, на сутки нет. Для кэша, где это недопустимо, выбирают write-through с версией значения или не кэшируют вовсе. Если ключей, связанных с сущностью, несколько (товар, товар в списке категории, товар в поиске), удаляют их все одним delete с несколькими аргументами или ведут набор ключей на сущность.

Удалять по маске KEYS product:* в проде нельзя: команда блокирует Redis на время обхода всех ключей. Для редких массовых чисток есть scan_iter(match="product:*", count=500) постранично и unlink, который освобождает память в фоне.

Конвейеры и транзакции

Каждая команда это сетевой круг. Десять GET подряд это десять кругов, MGET или конвейер это один:

async def mget_products(self, ids: list[int]) -> dict[int, CachedProduct]:
    values = await self.redis.mget([f"product:{i}" for i in ids])
    out: dict[int, CachedProduct] = {}
    for product_id, raw in zip(ids, values):
        if raw is None:
            continue
        try:
            out[product_id] = CachedProduct.model_validate_json(raw)
        except ValidationError:
            continue
    return out

redis.pipeline() отправляет разные команды одной пачкой и возвращает результаты списком; по умолчанию он оборачивает их в MULTI/EXEC, и они выполняются без вклинивания чужих команд, pipeline(transaction=False) даёт просто пачку. Это не транзакция в смысле отката: если третья команда упала, первые две уже применились. Так считают лимиты:

from datetime import UTC, datetime


async def allow(redis: Redis, user: str, limit: int) -> bool:
    key = f"rl:{user}:{datetime.now(UTC):%Y%m%d%H%M}"
    async with redis.pipeline() as pipe:
        pipe.incr(key)
        pipe.expire(key, 60)
        count, _ = await pipe.execute()
    return count <= limit

Результаты команд в конвейере приходят только из execute: вызовы pipe.incr и pipe.expire ничего не возвращают и не требуют await. Окно здесь фиксированное по минутам; скользящее окно делают сортированным множеством с отметками времени.

Блокировки: владелец и Lua

Распределённая блокировка на одном Redis это SET key owner NX PX ttl, а освобождение обязано проверять владельца, иначе процесс, который провисел дольше TTL, снимет чужую блокировку. Проверка и удаление должны быть одной атомарной операцией, то есть Lua-скриптом, и redis-py всё это уже умеет:

from redis.exceptions import LockError


async def run_nightly(redis: Redis) -> None:
    lock = redis.lock("lock:nightly-report", timeout=300, blocking=False)
    if not await lock.acquire():
        log.info("nightly report already running elsewhere")
        return
    try:
        await build_report(extend=lambda: lock.extend(300))
    finally:
        try:
            await lock.release()
        except LockError:
            log.warning("lock expired before release: report took longer than ttl")

Объект Lock сам генерирует случайный токен владельца на захват и освобождает ключ Lua-скриптом, который сверяет токен; release чужой или истёкшей блокировки даёт LockError, а не молча снимает её. timeout больше ожидаемого времени работы с запасом, а долгая задача продлевает его через extend. Такая блокировка защищает от двойного запуска задачи по расписанию на нескольких экземплярах; для денег она не годится: при переключении мастера Redis блокировка может пропасть, а значит, идемпотентность операции обеспечивают в базе, а не в Redis. Алгоритм Redlock на нескольких независимых узлах реализуют отдельные библиотеки, и спор о его гарантиях стоит прочитать до того, как на него опираться.

Сессии

Сессия в Redis это значение по ключу sess:<id> с TTL, а идентификатор живёт в cookie. Скользящее продление делает одна команда:

import secrets

SESSION_TTL = 30 * 60


class Session(BaseModel):
    user_id: str
    roles: list[str]


async def create_session(redis: Redis, session: Session) -> str:
    sid = secrets.token_urlsafe(32)
    await redis.set(f"sess:{sid}", session.model_dump_json(), ex=SESSION_TTL)
    return sid


async def load_session(redis: Redis, sid: str) -> Session | None:
    raw = await redis.getex(f"sess:{sid}", ex=SESSION_TTL)
    return Session.model_validate_json(raw) if raw else None

GETEX читает и продлевает TTL за один круг. Идентификатор сессии генерируют криптографически (secrets.token_urlsafe(32)), в cookie ставят HttpOnly, Secure и SameSite. Отзыв сессии это delete, и в этом главное преимущество перед JWT без состояния: выход и блокировка пользователя действуют мгновенно. Для сервисов с JWT Redis остаётся местом для чёрного списка отозванных токенов с TTL до их истечения.

Когда Redis лежит

Поведение при недоступности выбирают для каждого применения отдельно, и это решение записывают в код явно.

  • Кэш отказывает открыто. Ошибка Redis это промах: лог, метрика, запрос в базу. Чтобы при зависшем Redis не платить таймаут на каждый запрос, перед кэшем ставят размыкатель (aiobreaker, pybreaker или свой счётчик ошибок): после серии ошибок кэш отключается на 10–30 секунд, и запросы идут в базу сразу. База в этот момент получает весь трафик, поэтому её запас прочности на холодный кэш проверяют заранее.
  • Лимиты и блокировки отказывают по решению продукта. Пропустить запрос без проверки лимита или отказать всем? Для публичного API обычно пропускают и пишут метрику, для дорогих операций отказывают.
  • Сессии отказывают закрыто. Без Redis пользователь не аутентифицирован, и это 503 с коротким таймаутом, а не бесконечное ожидание. Поэтому Redis под сессии держат с репликой и Sentinel или Cluster.

Проверка здоровья отражает это: кэш это «деградация», сессии это «не готов». Короткие таймауты и один повтор из первого раздела здесь и работают: сервис замечает проблему за сотни миллисекунд, а не за секунды. Отдельный сигнал — BusyLoadingError: Redis поднялся и грузит снимок в память, несколько секунд он отвечает отказом, и это тоже промах, а не авария.

Sentinel и Cluster

Sentinel([("sentinel-1", 26379), ("sentinel-2", 26379)]).master_for("mymaster", socket_timeout=0.3) сам находит мастера и переподключается после переключения; RedisCluster.from_url(...) работает с кластером и маршрутизирует команды по слотам. Два следствия для кода. Во-первых, команды над несколькими ключами (MGET, MULTI, Lua с двумя ключами) в кластере работают только внутри одного слота, поэтому связанные ключи получают общий хеш-тег: cart:{42}:items и cart:{42}:total лягут вместе. Во-вторых, чтение с реплик (read_from_replicas=True) ускоряет кэш, но может отдать значение, которое мастер уже удалил, и для сессий и блокировок его не включают.

Наблюдаемость

RedisInstrumentor().instrument() из opentelemetry-instrumentation-redis добавляет каждой команде отрезок трассировки с именем команды и ключом. Статистики пула, как в клиентах других языков, у redis-py нет: число занятых соединений считают сами, а client_name в настройках клиента помогает найти соединения сервиса в CLIENT LIST на стороне Redis. Для самого кэша считают свои метрики: попадания и промахи по имени кэша и время ответа; без доли попаданий нельзя сказать, нужен ли кэш вообще.

Тесты

Логику кэша тестируют на fakeredis: fakeredis.aioredis.FakeRedis() это Redis внутри процесса с тем же интерфейсом, стартует за миллисекунды и понимает TTL. Реализацию против настоящего сервера проверяет контейнер:

import pytest
from testcontainers.redis import RedisContainer


@pytest.fixture(scope="session")
def redis_url() -> str:
    with RedisContainer("redis:7") as container:
        yield f"redis://{container.get_container_host_ip()}:{container.get_exposed_port(6379)}/0"

Что проверяют: промах идёт в базу один раз при ста одновременных вызовах (asyncio.gather на сто get и счётчик вызовов загрузчика), ошибка Redis не ломает чтение (подменить адрес на закрытый порт), инвалидация после обновления, лимит отсекает ровно на limit + 1, чужой владелец не снимает блокировку.

Дополнительно: при первом чтении можно пропустить

Глубже: горячие ключи и большие значениярасширенное

Один ключ, который читают десятки тысяч раз в секунду (главная страница, конфигурация), упирается в одно ядро Redis и в одну сетевую карту узла кластера. Лечат его локальным кэшем в памяти процесса на секунды перед Redis: cachetools.TTLCache плюс тот же словарь inflight. Большое значение (мегабайты JSON) тормозит всех: Redis однопоточный, и пока он отдаёт мегабайт одному клиенту, остальные ждут. Правило: значения в килобайтах, большее режут на части, сжимают или не кладут в кэш вообще. Удаление большого ключа тоже блокирует, поэтому unlink вместо delete.

Коротко

  • Один клиент на процесс; таймауты в сотни миллисекунд вместо пяти секунд, один повтор вместо десяти с явным retry_on_error, пул 10–20 вместо 100.
  • None из get это промах; любая ошибка кэша и ValidationError тоже промах с логом, а не 500.
  • Cache-aside со словарём Future на ключ, TTL с разбросом и кэшем отсутствия закрывает лавину промахов.
  • Инвалидация удалением после коммита; гонку чтения закрывает короткий TTL; KEYS в проде запрещён, вместо него scan_iter и unlink.
  • mget и конвейеры вместо циклов команд; pipeline() по умолчанию атомарная пачка MULTI/EXEC, а не транзакция с откатом; результаты только из execute.
  • Блокировка: redis.lock с токеном владельца и Lua-освобождением, extend для долгих задач, LockError вместо молчаливого снятия; деньги защищает идемпотентность в базе, а не Redis.
  • Сессии по ключу с TTL и getex, идентификатор из secrets, отзыв через delete.
  • При падении Redis кэш отказывает открыто и за размыкателем, сессии закрыто с коротким таймаутом; BusyLoadingError после перезапуска — тоже промах.
  • В кластере связанные ключи с общим хеш-тегом {id}, чтение с реплик только для кэша.
  • opentelemetry-instrumentation-redis и свои метрики попаданий для наблюдения, fakeredis для юнит-тестов, контейнер для реализации.

Что почитать дальше