Профессиональный Генератор Кода

Rate limiting middleware для booking-api (FastAPI + Redis, sliding window)

1. Подход

Допущения:

1. user_id берётся только из проверенного JWT. Если брать его из неподписанного токена, атакующий может подставлять случайные user_id и обходить любые лимиты. Поэтому middleware не парсит JWT сам. Он получает IdentityResolver (callable). Resolver по умолчанию читает request.state.user_id, который выставляет auth-middleware, стоящий выше по стеку. Пример resolver'а с проверкой подписи приведён в разделе «Использование».
2. Один ключ на запрос. Авторизованный запрос ограничивается по user_id (user_limit), анонимный — по IP (ip_limit). Лимиты задаются раздельно для каждой группы эндпоинтов. В ТЗ даны только значения для пользователей, поэтому IP-лимиты по умолчанию я взял строже: bookings 10/10s, search 50/10s, default 30/60s.
3. Реальный IP за ingress. Для него используется trusted_proxy_hops: сколько доверенных прокси дописывают X-Forwarded-For. По умолчанию 0, то есть XFF игнорируется, потому что он подделывается тривиально.
4. Отклонённые запросы не увеличивают счётчик. Иначе клиент, который продолжает слать запросы, заблокирует себя навсегда.

Дизайн:

• Алгоритм — sliding window counter. Хранятся два счётчика (текущее и предыдущее окно), предыдущий учитывается с весом. Это даёт O(1) памяти на ключ, в отличие от sliding log на ZSET, который хранит O(limit) элементов и дорог при 5000 RPS.
• Состояние в Redis. Оно лежит в одном hash-ключе на идентичность. Lua-скрипт атомарно читает, проверяет и инкрементирует счётчик за один RTT через EVALSHA.
• Время берётся из redis.call('TIME'). Так часы 6–20 подов не рассинхронизируются между собой. Один ключ на скрипт делает решение совместимым с Redis Cluster.
• Middleware написан на чистом ASGI, без BaseHTTPMiddleware. Это меньше накладных расходов и нет проблем со стримингом.
• Fail-open с circuit breaker. После сбоя Redis middleware на breaker_cooldown_seconds перестаёт к нему ходить. Это убирает лишние таймауты на каждом запросе и шторм WARNING-логов на 5000 RPS.

Структура пакета:

booking_api/rate_limit/
├── __init__.py
├── config.py      # Pydantic Settings
├── limiter.py     # Lua + SlidingWindowRateLimiter (чистая логика + Redis)
└── middleware.py  # ASGI middleware: идентификация, whitelist, fail-open, 429

2. Код

config.py

"""Конфигурация rate limiting, загружаемая из переменных окружения."""

from __future__ import annotations

from typing import Self

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    IPvAnyNetwork,
    PositiveInt,
    field_validator,
    model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict


class Limit(BaseModel):
    """Лимит «не более `requests` запросов за `window_seconds` секунд»."""

    model_config = ConfigDict(frozen=True)

    requests: PositiveInt
    window_seconds: PositiveInt


class LimitPolicy(BaseModel):
    """Именованная пара лимитов: по user_id (авторизованные) и по IP (анонимные).

    Имя входит в ключ Redis, поэтому ограничено безопасным набором символов.
    """

    model_config = ConfigDict(frozen=True)

    name: str = Field(pattern=r"^[a-z0-9_-]{1,32}$")
    user_limit: Limit
    ip_limit: Limit


class EndpointGroup(LimitPolicy):
    """Политика для путей с заданным префиксом.

    `path_prefix` принимает как `/bookings`, так и `/bookings/*`. Совпадение
    проверяется по границе сегмента: `/bookings` матчит `/bookings/42`,
    но не `/bookingsX`.
    """

    path_prefix: str

    @field_validator("path_prefix")
    @classmethod
    def _normalize_prefix(cls, value: str) -> str:
        prefix = value.removesuffix("*").rstrip("/")
        if not prefix.startswith("/"):
            raise ValueError(
                "path_prefix must start with '/' and must not be a catch-all "
                "(use default_policy for that)"
            )
        return prefix

    def matches(self, path: str) -> bool:
        """Проверяет, относится ли путь к группе (без аллокаций на горячем пути)."""
        prefix = self.path_prefix
        return path.startswith(prefix) and (
            len(path) == len(prefix) or path[len(prefix)] == "/"
        )


_DEFAULT_GROUPS: tuple[EndpointGroup, ...] = (
    EndpointGroup(
        name="bookings",
        path_prefix="/bookings/*",
        user_limit=Limit(requests=20, window_seconds=10),
        ip_limit=Limit(requests=10, window_seconds=10),
    ),
    EndpointGroup(
        name="search",
        path_prefix="/search/*",
        user_limit=Limit(requests=100, window_seconds=10),
        ip_limit=Limit(requests=50, window_seconds=10),
    ),
)

_DEFAULT_POLICY = LimitPolicy(
    name="default",
    user_limit=Limit(requests=60, window_seconds=60),
    ip_limit=Limit(requests=30, window_seconds=60),
)


class RateLimitSettings(BaseSettings):
    """Настройки rate limiting.

    Сложные поля задаются JSON-строкой в env, например::

        RATE_LIMIT_ENABLED=true
        RATE_LIMIT_TRUSTED_PROXY_HOPS=1
        RATE_LIMIT_WHITELIST='["10.0.0.0/8", "192.168.10.5"]'
        RATE_LIMIT_GROUPS='[{"name": "bookings", "path_prefix": "/bookings/*",
            "user_limit": {"requests": 20, "window_seconds": 10},
            "ip_limit": {"requests": 10, "window_seconds": 10}}]'

    Attributes:
        enabled: Глобальный рубильник (ручное отключение rate limiting).
        key_prefix: Префикс ключей в Redis.
        redis_timeout_seconds: Жёсткий бюджет времени на обращение к Redis.
        breaker_cooldown_seconds: Сколько секунд не ходить в Redis после сбоя.
        trusted_proxy_hops: Число доверенных прокси, дописывающих X-Forwarded-For.
        whitelist: IP/CIDR, освобождённые от лимитов.
        groups: Группы эндпоинтов; при пересечении побеждает самый длинный префикс.
        default_policy: Политика для путей, не попавших ни в одну группу.
    """

    model_config = SettingsConfigDict(env_prefix="RATE_LIMIT_", frozen=True)

    enabled: bool = True
    key_prefix: str = Field(default="rl", pattern=r"^[a-zA-Z0-9_:-]{1,32}$")
    redis_timeout_seconds: float = Field(default=0.1, gt=0, le=5)
    breaker_cooldown_seconds: float = Field(default=5.0, ge=0)
    trusted_proxy_hops: int = Field(default=0, ge=0, le=10)
    whitelist: tuple[IPvAnyNetwork, ...] = ()
    groups: tuple[EndpointGroup, ...] = _DEFAULT_GROUPS
    default_policy: LimitPolicy = _DEFAULT_POLICY

    @model_validator(mode="after")
    def _check_unique_policy_names(self) -> Self:
        names = [group.name for group in self.groups]
        names.append(self.default_policy.name)
        if len(names) != len(set(names)):
            raise ValueError(f"policy names must be unique, got {names}")
        return self

limiter.py

"""Sliding window counter rate limiter на Redis (атомарно через Lua)."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Final

import asyncio
from redis.asyncio import Redis
from redis.exceptions import RedisError

from .config import Limit

# Алгоритм sliding window counter.
# Состояние: один hash на идентичность: w = индекс текущего окна,
# c = счётчик текущего окна, p = счётчик предыдущего окна.
# Оценка числа запросов за последние `window` секунд:
#     estimated = p * (window - elapsed) / window + c
# Время берётся из Redis TIME (единые часы для всех подов). В Redis 7
# действует effects replication, поэтому недетерминированный TIME в скрипте
# допустим. Один ключ на вызов делает скрипт совместимым с Redis Cluster.
# Возвращает {allowed (0|1), remaining, retry_after_ms}.
_SLIDING_WINDOW_LUA: Final = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])

local t = redis.call('TIME')
local now = tonumber(t[1]) + tonumber(t[2]) / 1000000
local idx = math.floor(now / window)
local elapsed = now - idx * window

local state = redis.call('HMGET', key, 'w', 'c', 'p')
local w = tonumber(state[1])
local cur = tonumber(state[2]) or 0
local prev = tonumber(state[3]) or 0

if w ~= idx then
  if w == idx - 1 then prev = cur else prev = 0 end
  cur = 0
end

local weight = (window - elapsed) / window
if prev * weight + cur + 1 > limit then
  -- Ищем ближайший момент, когда ещё один запрос уложится в лимит.
  local wait
  if cur + 1 > limit then
    -- Текущее окно исчерпано само по себе: ждём его конца, затем пока
    -- вес cur (ставшего prev) не упадёт до (limit - 1) / cur.
    wait = (window - elapsed) + window * (1 - (limit - 1) / cur)
  else
    -- Хватит затухания предыдущего окна (здесь prev > 0 гарантированно).
    wait = window * (1 - (limit - cur - 1) / prev) - elapsed
  end
  if wait < 0.001 then wait = 0.001 end
  return {0, 0, math.ceil(wait * 1000)}
end

cur = cur + 1
redis.call('HSET', key, 'w', idx, 'c', cur, 'p', prev)
redis.call('PEXPIRE', key, window * 2000)
local remaining = math.floor(limit - (prev * weight + cur))
if remaining < 0 then remaining = 0 end
return {1, remaining, 0}
"""


class RateLimiterBackendError(Exception):
    """Хранилище лимитов недоступно или вернуло некорректный ответ."""


@dataclass(frozen=True, slots=True)
class RateLimitResult:
    """Результат проверки лимита.

    Attributes:
        allowed: Разрешён ли запрос.
        limit: Действующий лимит (для заголовка X-RateLimit-Limit).
        remaining: Сколько запросов ещё доступно в скользящем окне.
        retry_after_seconds: Через сколько секунд повторить (0, если allowed).
    """

    allowed: bool
    limit: int
    remaining: int
    retry_after_seconds: int


class SlidingWindowRateLimiter:
    """Распределённый rate limiter на Redis (sliding window counter).

    Каждый вызов `hit` — ровно один EVALSHA (один RTT). redis-py сам
    перезагружает скрипт при NoScriptError, например после рестарта Redis.

    Example:
        >>> limiter = SlidingWindowRateLimiter(redis, timeout_seconds=0.1)
        >>> result = await limiter.hit("rl:search:ip:1.2.3.4", Limit(requests=100, window_seconds=10))
    """

    def __init__(self, redis: Redis, *, timeout_seconds: float) -> None:
        """Инициализирует limiter.

        Args:
            redis: Асинхронный клиент Redis (переиспользуемый, с пулом соединений).
            timeout_seconds: Жёсткий бюджет на вызов, включая ожидание соединения из пула.
        """
        self._script = redis.register_script(_SLIDING_WINDOW_LUA)
        self._timeout = timeout_seconds

    async def hit(self, key: str, limit: Limit) -> RateLimitResult:
        """Атомарно проверяет лимит и, если запрос разрешён, учитывает его.

        Args:
            key: Ключ Redis для идентичности (user/IP + группа).
            limit: Применяемый лимит.

        Returns:
            Результат проверки.

        Raises:
            RateLimiterBackendError: Redis недоступен, превышен таймаут
                или получен неожиданный ответ.
        """
        try:
            async with asyncio.timeout(self._timeout):
                raw = await self._script(
                    keys=[key], args=[limit.requests, limit.window_seconds]
                )
        # TimeoutError (asyncio.timeout) и сетевые ошибки — подклассы OSError;
        # ConnectionError/TimeoutError/ResponseError из redis — подклассы RedisError.
        except (RedisError, OSError) as exc:
            raise RateLimiterBackendError(f"{type(exc).__name__}: {exc}") from exc
        return _parse_result(raw, limit.requests)


def _parse_result(raw: Any, limit: int) -> RateLimitResult:
    try:
        allowed, remaining, retry_after_ms = (int(value) for value in raw)
    except (TypeError, ValueError) as exc:
        raise RateLimiterBackendError(f"unexpected script reply: {raw!r}") from exc
    return RateLimitResult(
        allowed=bool(allowed),
        limit=limit,
        remaining=max(remaining, 0),
        # Ceil до целых секунд, минимум 1: Retry-After: 0 провоцирует мгновенный ретрай.
        retry_after_seconds=max(1, -(-retry_after_ms // 1000)) if not allowed else 0,
    )

middleware.py

"""ASGI middleware ограничения частоты запросов."""

from __future__ import annotations

import logging
import time
from collections.abc import Callable
from dataclasses import dataclass
from ipaddress import ip_address

from redis.asyncio import Redis
from starlette.datastructures import MutableHeaders
from starlette.responses import JSONResponse
from starlette.types import ASGIApp, Message, Receive, Scope, Send

from .config import Limit, LimitPolicy, RateLimitSettings
from .limiter import RateLimiterBackendError, RateLimitResult, SlidingWindowRateLimiter

logger = logging.getLogger(__name__)

IdentityResolver = Callable[[Scope], str | None]
"""Возвращает проверенный user_id для запроса или None для анонимного."""

MetricsHook = Callable[[str, str], None]


def user_id_from_state(scope: Scope) -> str | None:
    """Resolver по умолчанию: берёт `request.state.user_id`.

    Значение должен выставить auth-middleware, который выполняется раньше и
    проверяет подпись JWT. `Request.state` хранится в `scope["state"]`.
    """
    state = scope.get("state")
    if not state:
        return None
    user_id = state.get("user_id")
    return str(user_id) if user_id is not None else None


@dataclass(frozen=True, slots=True)
class RateLimitHooks:
    """Хуки для метрик, не привязанные к конкретному клиенту метрик.

    Attributes:
        on_rejected: Вызывается при ответе 429, аргументы — (policy, key_kind),
            где key_kind — "user" или "ip".
        on_backend_error: Вызывается при fail-open, аргументы — (policy, error_type).
    """

    on_rejected: MetricsHook | None = None
    on_backend_error: MetricsHook | None = None


class _FailOpenBreaker:
    """Простейший circuit breaker: после сбоя не ходим в Redis `cooldown` секунд.

    Не даёт каждому из тысяч RPS ждать таймаут мёртвого Redis и спамить WARNING.
    Состояние локально для процесса, что достаточно для fail-open.
    """

    __slots__ = ("_clock", "_cooldown", "_open_until")

    def __init__(self, cooldown: float, clock: Callable[[], float] = time.monotonic) -> None:
        self._cooldown = cooldown
        self._clock = clock
        self._open_until = 0.0

    def is_open(self) -> bool:
        return self._clock() < self._open_until

    def trip(self) -> None:
        self._open_until = self._clock() + self._cooldown


class RateLimitMiddleware:
    """Rate limiting по user_id (авторизованные) или IP (анонимные) на Redis.

    Порядок обработки: не-HTTP / выключено -> whitelist -> выбор политики по пути
    -> проверка в Redis -> 429 либо проксирование с заголовками X-RateLimit-*.
    При недоступности Redis запрос пропускается (fail-open) с WARNING-логом.

    Example:
        >>> app.add_middleware(RateLimitMiddleware, redis=redis, settings=RateLimitSettings())
    """

    def __init__(
        self,
        app: ASGIApp,
        *,
        redis: Redis,
        settings: RateLimitSettings,
        identity_resolver: IdentityResolver = user_id_from_state,
        hooks: RateLimitHooks | None = None,
    ) -> None:
        """Инициализирует middleware.

        Args:
            app: Следующее ASGI-приложение.
            redis: Асинхронный клиент Redis.
            settings: Настройки лимитов.
            identity_resolver: Извлекает проверенный user_id из scope.
            hooks: Хуки метрик.
        """
        self._app = app
        self._settings = settings
        self._limiter = SlidingWindowRateLimiter(
            redis, timeout_seconds=settings.redis_timeout_seconds
        )
        self._resolve_identity = identity_resolver
        self._hooks = hooks or RateLimitHooks()
        self._breaker = _FailOpenBreaker(settings.breaker_cooldown_seconds)
        # Самый длинный префикс проверяется первым (самое специфичное правило).
        self._groups = sorted(settings.groups, key=lambda g: len(g.path_prefix), reverse=True)
        self._whitelist = settings.whitelist

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http" or not self._settings.enabled:
            await self._app(scope, receive, send)
            return

        client_ip = self._client_ip(scope)
        if self._is_whitelisted(client_ip):
            await self._app(scope, receive, send)
            return

        path: str = scope["path"]
        policy = self._resolve_policy(path)
        user_id = self._resolve_identity(scope)
        if user_id:
            kind, identity, limit = "user", user_id, policy.user_limit
        else:
            kind, identity, limit = "ip", client_ip or "unknown", policy.ip_limit

        result = await self._check(policy, kind, identity, limit, path)
        if result is None:  # fail-open
            await self._app(scope, receive, send)
        elif not result.allowed:
            _emit(self._hooks.on_rejected, policy.name, kind)
            await self._reject(result)(scope, receive, send)
        else:
            await self._app(scope, receive, _with_rate_limit_headers(send, result))

    async def _check(
        self, policy: LimitPolicy, kind: str, identity: str, limit: Limit, path: str
    ) -> RateLimitResult | None:
        """Проверяет лимит; возвращает None, если нужно пропустить запрос (fail-open)."""
        if self._breaker.is_open():
            return None
        key = f"{self._settings.key_prefix}:{policy.name}:{kind}:{identity}"
        try:
            return await self._limiter.hit(key, limit)
        except RateLimiterBackendError as exc:
            self._breaker.trip()
            error_type = type(exc.__cause__ or exc).__name__
            logger.warning(
                "Rate limiter fail-open: endpoint=%s policy=%s reason=%s "
                "(limits disabled for %.1fs)",
                path,
                policy.name,
                exc,
                self._settings.breaker_cooldown_seconds,
                exc_info=exc,
                extra={
                    "event": "rate_limit_fail_open",
                    "endpoint": path,
                    "policy": policy.name,
                    "error_type": error_type,
                },
            )
            _emit(self._hooks.on_backend_error, policy.name, error_type)
            return None

    def _resolve_policy(self, path: str) -> LimitPolicy:
        for group in self._groups:
            if group.matches(path):
                return group
        return self._settings.default_policy

    def _client_ip(self, scope: Scope) -> str | None:
        """Определяет IP клиента с учётом доверенных прокси.

        При trusted_proxy_hops=N берётся N-й адрес справа в X-Forwarded-For:
        правые N записей добавлены нашими прокси, левее — данные клиента,
        которым доверять нельзя.
        """
        hops = self._settings.trusted_proxy_hops
        if hops:
            for name, value in scope["headers"]:
                if name == b"x-forwarded-for":
                    parts = [p.strip() for p in value.decode("latin-1").split(",") if p.strip()]
                    if parts:
                        return parts[max(len(parts) - hops, 0)]
                    break
        client = scope.get("client")
        return client[0] if client else None

    def _is_whitelisted(self, client_ip: str | None) -> bool:
        if not self._whitelist or client_ip is None:
            return False
        try:
            address = ip_address(client_ip)
        except ValueError:
            return False
        return any(address in network for network in self._whitelist)

    @staticmethod
    def _reject(result: RateLimitResult) -> JSONResponse:
        return JSONResponse(
            {"detail": "Too Many Requests"},
            status_code=429,
            headers={
                "Retry-After": str(result.retry_after_seconds),
                "X-RateLimit-Limit": str(result.limit),
                "X-RateLimit-Remaining": "0",
            },
        )


def _with_rate_limit_headers(send: Send, result: RateLimitResult) -> Send:
    """Оборачивает send, добавляя X-RateLimit-* в успешный ответ."""

    async def wrapped(message: Message) -> None:
        if message["type"] == "http.response.start":
            headers = MutableHeaders(scope=message)
            headers.append("X-RateLimit-Limit", str(result.limit))
            headers.append("X-RateLimit-Remaining", str(result.remaining))
        await send(message)

    return wrapped


def _emit(hook: MetricsHook | None, *args: str) -> None:
    """Вызывает хук метрик; сбой метрик не должен ронять запрос."""
    if hook is None:
        return
    try:
        hook(*args)
    except Exception:
        logger.exception("Rate limit metrics hook failed")

__init__.py

from .config import EndpointGroup, Limit, LimitPolicy, RateLimitSettings
from .limiter import RateLimiterBackendError, RateLimitResult, SlidingWindowRateLimiter
from .middleware import RateLimitHooks, RateLimitMiddleware, user_id_from_state

__all__ = [
    "EndpointGroup", "Limit", "LimitPolicy", "RateLimitHooks", "RateLimitMiddleware",
    "RateLimitResult", "RateLimitSettings", "RateLimiterBackendError",
    "SlidingWindowRateLimiter", "user_id_from_state",
]

3. Тесты

Требуются fakeredis[lua]: Lua-скрипты выполняются через lupa. Во всех тестах окна по 3600 с, чтобы граница окна гарантированно не попадала на время теста.

# tests/test_rate_limit.py
from __future__ import annotations

import logging
from typing import Any

import pytest
from fakeredis import FakeAsyncRedis, FakeServer
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from starlette.types import Scope

from booking_api.rate_limit import (
    EndpointGroup, Limit, LimitPolicy, RateLimitHooks, RateLimitMiddleware, RateLimitSettings,
)

pytestmark = pytest.mark.asyncio

HOUR = 3600
PUBLIC_IP = "203.0.113.10"


def header_identity(scope: Scope) -> str | None:
    """Тестовый resolver: user_id из заголовка (в проде — только проверенный JWT)."""
    for name, value in scope["headers"]:
        if name == b"x-test-user":
            return str(value.decode())
    return None


def make_settings(**overrides: Any) -> RateLimitSettings:
    params: dict[str, Any] = {
        "groups": (
            EndpointGroup(
                name="bookings", path_prefix="/bookings/*",
                user_limit=Limit(requests=2, window_seconds=HOUR),
                ip_limit=Limit(requests=3, window_seconds=HOUR),
            ),
        ),
        "default_policy": LimitPolicy(
            name="default",
            user_limit=Limit(requests=100, window_seconds=HOUR),
            ip_limit=Limit(requests=100, window_seconds=HOUR),
        ),
        "whitelist": ("10.0.0.0/8",),
        "breaker_cooldown_seconds": 0,
    }
    params.update(overrides)
    return RateLimitSettings(**params)


def build_app(redis: FakeAsyncRedis, hooks: RateLimitHooks | None = None) -> FastAPI:
    app = FastAPI()

    @app.get("/bookings/{booking_id}")
    async def get_booking(booking_id: int) -> dict[str, int]:
        return {"id": booking_id}

    app.add_middleware(
        RateLimitMiddleware, redis=redis, settings=make_settings(),
        identity_resolver=header_identity, hooks=hooks,
    )
    return app


def client_for(app: FastAPI, ip: str = PUBLIC_IP) -> AsyncClient:
    return AsyncClient(transport=ASGITransport(app=app, client=(ip, 50000)), base_url="http://t")


@pytest.fixture
def redis() -> FakeAsyncRedis:
    return FakeAsyncRedis(server=FakeServer())


async def test_requests_within_limit_pass_with_decreasing_remaining(redis: FakeAsyncRedis) -> None:
    async with client_for(build_app(redis)) as client:
        responses = [await client.get("/bookings/1") for _ in range(3)]

    assert [r.status_code for r in responses] == [200, 200, 200]
    assert [r.headers["X-RateLimit-Remaining"] for r in responses] == ["2", "1", "0"]
    assert responses[0].headers["X-RateLimit-Limit"] == "3"


async def test_exceeding_limit_returns_429_with_retry_after(redis: FakeAsyncRedis) -> None:
    rejected: list[tuple[str, str]] = []
    hooks = RateLimitHooks(on_rejected=lambda policy, kind: rejected.append((policy, kind)))

    async with client_for(build_app(redis, hooks)) as client:
        for _ in range(3):
            assert (await client.get("/bookings/1")).status_code == 200
        response = await client.get("/bookings/1")

    assert response.status_code == 429
    assert response.json() == {"detail": "Too Many Requests"}
    assert response.headers["X-RateLimit-Remaining"] == "0"
    # limit=3, cur=3, prev=0: ждать (window - elapsed) + window/3, т.е. от 1200 до 4800 с.
    assert 1200 <= int(response.headers["Retry-After"]) <= 4800
    assert rejected == [("bookings", "ip")]


async def test_whitelisted_ip_is_never_limited(redis: FakeAsyncRedis) -> None:
    async with client_for(build_app(redis), ip="10.1.2.3") as client:
        responses = [await client.get("/bookings/1") for _ in range(10)]

    assert all(r.status_code == 200 for r in responses)
    assert "X-RateLimit-Remaining" not in responses[-1].headers
    assert await redis.dbsize() == 0


async def test_fail_open_when_redis_unavailable(caplog: pytest.LogCaptureFixture) -> None:
    server = FakeServer()
    server.connected = False  # любая команда -> redis.ConnectionError
    errors: list[tuple[str, str]] = []
    hooks = RateLimitHooks(on_backend_error=lambda policy, err: errors.append((policy, err)))

    with caplog.at_level(logging.WARNING, logger="booking_api.rate_limit.middleware"):
        async with client_for(build_app(FakeAsyncRedis(server=server), hooks)) as client:
            response = await client.get("/bookings/7")

    assert response.status_code == 200
    assert "X-RateLimit-Remaining" not in response.headers
    assert errors == [("bookings", "ConnectionError")]
    [record] = [r for r in caplog.records if getattr(r, "event", None) == "rate_limit_fail_open"]
    assert record.levelno == logging.WARNING
    assert "/bookings/7" in record.getMessage()
    assert record.exc_info is not None  # исключение не проглочено молча


async def test_user_and_ip_limits_are_independent(redis: FakeAsyncRedis) -> None:
    alice = {"X-Test-User": "alice"}
    async with client_for(build_app(redis)) as client:
        assert [(await client.get("/bookings/1", headers=alice)).status_code
                for _ in range(3)] == [200, 200, 429]  # user_limit = 2
        # Тот же IP, но анонимно: свой IP-бакет (limit 3) не тронут.
        anon = await client.get("/bookings/1")
        # Другой пользователь с того же IP: свой бакет.
        bob = await client.get("/bookings/1", headers={"X-Test-User": "bob"})

    assert anon.status_code == 200 and anon.headers["X-RateLimit-Remaining"] == "2"
    assert bob.status_code == 200 and bob.headers["X-RateLimit-Limit"] == "2"

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

# pyproject.toml (фрагмент)
dependencies = ["fastapi==0.110.*", "redis>=5.0", "pydantic-settings>=2.2", "pyjwt>=2.8"]
[project.optional-dependencies]
test = ["pytest>=8", "pytest-asyncio>=0.23", "httpx>=0.27", "fakeredis[lua]>=2.23"]

types-redis не ставить: redis-py ≥5 поставляется с собственными типами (py.typed), а старые стабы конфликтуют с ними под mypy --strict.

# booking_api/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import jwt
from fastapi import FastAPI
from prometheus_client import Counter
from redis.asyncio import Redis
from starlette.types import Scope

from booking_api.rate_limit import RateLimitHooks, RateLimitMiddleware, RateLimitSettings

REJECTED = Counter("rate_limit_rejected_total", "429 responses", ["policy", "key_kind"])
FAIL_OPEN = Counter("rate_limit_fail_open_total", "Redis failures", ["policy", "error"])

redis = Redis.from_url(
    "redis://redis:6379/0",
    socket_timeout=0.1, socket_connect_timeout=0.2, max_connections=200,
)


def jwt_identity(scope: Scope) -> str | None:
    """Проверяет подпись; невалидный токен -> анонимный (лимит по IP)."""
    for name, value in scope["headers"]:
        if name == b"authorization" and value.startswith(b"Bearer "):
            try:
                claims = jwt.decode(value[7:].decode(), PUBLIC_KEY, algorithms=["RS256"])
            except jwt.PyJWTError:
                return None
            user_id = claims.get("user_id")
            return str(user_id) if user_id is not None else None
    return None


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    yield
    await redis.aclose()


app = FastAPI(lifespan=lifespan)
app.add_middleware(
    RateLimitMiddleware,
    redis=redis,
    settings=RateLimitSettings(),
    identity_resolver=jwt_identity,  # или user_id_from_state, если auth-middleware уже есть
    hooks=RateLimitHooks(
        on_rejected=lambda p, k: REJECTED.labels(p, k).inc(),
        on_backend_error=lambda p, e: FAIL_OPEN.labels(p, e).inc(),
    ),
)

Если используете user_id_from_state, auth-middleware добавляйте после RateLimitMiddleware: в Starlette последний добавленный middleware становится внешним и выполняется первым.

Алерт: increase(rate_limit_fail_open_total[5m]) > 0 или по логам с event=rate_limit_fail_open.

Проверки:

pytest -q && mypy --strict booking_api && ruff check booking_api tests

5. Возможные улучшения

1. Безопасность IP-ключей.
   • Ошибка в trusted_proxy_hops позволяет подделать IP из whitelist через XFF. Эту настройку стоит покрыть интеграционным тестом с реальным ingress.
   • Для IPv6 лучше агрегировать ключ до /64: клиент с одним префиксом легко генерирует миллионы адресов и обходит лимит.
   • Бакет "unknown" (нет scope["client"]) общий для всех таких запросов. Если это встречается в вашем окружении, такие запросы стоит отклонять.
2. Нагрузка и отказоустойчивость Redis.
   • На 5000 RPS (при 20 подах — до ~20k) один EVALSHA на запрос нормально переносит один инстанс Redis 7. Узким местом станет RTT, поэтому держите Redis в той же AZ.
   • Дальнейшие шаги: Redis Cluster (скрипт уже односоставной по ключам) и выделенный инстанс, чтобы не делить его с кэшем.
   • Breaker сейчас переходит «открыт → закрыт» сразу, и после cooldown все корутины одновременно идут в Redis. Имеет смысл добавить half-open состояние, пропускающее один пробный запрос.
3. Гибкость политик.
   • Если нужна защита от одного пользователя с множества IP и от одного IP с множеством аккаунтов, можно проверять оба ключа одновременно (user + IP) в одном Lua-вызове с двумя KEYS, используя hash-tag для cluster.
   • Отдельные лимиты для партнёров можно брать из claim тарифа в JWT.
   • Sliding window counter — это аппроксимация. Если нужна строгая точность на малых лимитах (например, 3 req/min для платежей), для этой группы можно добавить sliding log на ZSET.