Rate limiting: ограничение частоты запросов - IMI-DS

Привет! Rate limiting (ограничение частоты запросов) — это механизм защиты API от чрезмерного количества запросов от одного клиента. Он предотвращает DDoS-атаки, перебор паролей (brute-force) и злоупотребление ресурсами.

В этой статье мы разберём:

Что такое rate limiting и зачем он нужен
Реализация с SlowAPI
Реализация с fastapi-limiter
Кастомный middleware (без зависимостей)
Глобальные и per-route лимиты
Использование Redis для распределённых систем

Содержание

Что нужно знать перед началом
Основная часть
Что такое rate limiting
Реализация с SlowAPI
Реализация с fastapi-limiter
Кастомный middleware (без зависимостей)
fastapi-easylimiter
Полный пример с SlowAPI
Задачи для закрепления
Нюансы и подводные камни
В production используй Redis
Исключай health check из лимитов
Защита от brute-force
Обработка 429
Частые ошибки и как их избежать
Ошибка 1: In-memory в продакшене
Ошибка 2: Одинаковые лимиты для всех эндпоинтов
Ошибка 3: Отсутствие исключений для health check
Ошибка 4: Неправильная идентификация клиента
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание middleware и зависимостей

Совет: Rate limiting — это первая линия защиты API. Без него ваш сервис уязвим для злоупотреблений и атак .

Основная часть

Что такое rate limiting

Rate limiting — это ограничение количества запросов от одного клиента за определённый промежуток времени.

Зачем это нужно:

Защита от DDoS-атак
Предотвращение brute-force атак на логин
Контроль использования ресурсов
Обеспечение справедливого доступа для всех пользователей

Рекомендуемые лимиты:

Тип эндпоинта	Лимит
Auth (логин, регистрация)	5–10 запросов в минуту
POST (создание/обновление)	10–100 запросов в минуту
GET (чтение)	100–1000 запросов в минуту
Health check	Без ограничений

Совет: Для разных эндпоинтов используй разные лимиты.

Реализация с SlowAPI

SlowAPI — самая популярная библиотека для rate limiting в FastAPI. Она адаптирована из Flask-Limiter и работает как с синхронными, так и с асинхронными эндпоинтами.

Установка:

pip install slowapi

Базовый пример:

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

app = FastAPI()

# Создаём лимитер с ключом по IP
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/")
@limiter.limit("5/minute")
async def root(request: Request):
    return {"message": "Hello, World!"}

Совет: key_func=get_remote_address идентифицирует клиента по IP. Можно заменить на идентификатор пользователя .

Глобальный лимит для всех эндпоинтов:

from slowapi.middleware import SlowAPIMiddleware

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["100/minute"]  # Глобальный лимит
)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)

# Этот эндпоинт использует глобальный лимит
@app.get("/")
async def root(request: Request):
    return {"message": "Hello, World!"}

# Исключаем эндпоинт из глобального лимита
@app.get("/unlimited")
@limiter.exempt
async def unlimited(request: Request):
    return {"message": "I'm unlimited!"}

Совет: Используй @limiter.exempt для health check и других критических эндпоинтов .

Redis для продакшена:

from redis import Redis

redis_client = Redis(host="localhost", port=6379, db=0)

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
    default_limits=["100/hour"]
)

Совет: Redis необходим для распределённых систем, где несколько экземпляров приложения должны синхронизировать лимиты .

Реализация с fastapi-limiter

fastapi-limiter — альтернативная библиотека, построенная на основе pyrate-limiter. Она предоставляет простой dependency-based подход .

Установка:

pip install fastapi-limiter

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

import uvicorn
from fastapi import Depends, FastAPI
from pyrate_limiter import Duration, Limiter, Rate
from fastapi_limiter.depends import RateLimiter

app = FastAPI()

@app.get(
    "/",
    dependencies=[Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 5))))],
)
async def index():
    return {"msg": "Hello World"}

Несколько лимитеров в одном эндпоинте:

@app.get(
    "/multiple",
    dependencies=[
        Depends(RateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)))),
        Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 15)))),
    ],
)
async def multiple():
    return {"msg": "Multiple limiters"}

Совет: Более строгий лимитер должен быть первым.

Кастомный middleware (без зависимостей)

Если не хочется использовать сторонние библиотеки, можно реализовать свой простой rate limiter.

import time
from collections import defaultdict
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

# Храним историю запросов
request_log: defaultdict[str, list[float]] = defaultdict(list)

RATE = 5          # максимум запросов
PER_SECONDS = 60  # за 60 секунд

@app.middleware("http")
async def ip_rate_limiter(request: Request, call_next):
    ip = request.client.host
    now = time.time()

    # Очищаем старые записи
    request_log[ip] = [ts for ts in request_log[ip] if now - ts < PER_SECONDS]

    if len(request_log[ip]) >= RATE:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

    request_log[ip].append(now)
    response = await call_next(request)
    return response

Важно: Этот способ работает только в памяти одного экземпляра и не подходит для распределённых систем.

fastapi-easylimiter

fastapi-easylimiter — простая асинхронная библиотека с поддержкой Redis и in-memory .

Установка:

pip install fastapi-easylimiter

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

from fastapi import FastAPI
from fastapi_easylimiter import InMemoryBackend, RateLimiterMiddleware

app = FastAPI()

backend = InMemoryBackend()

rules = {
    "/api/users": {"limit": 1, "period": 2},
    "/api/": {"limit": 60, "period": 60},
}

app.add_middleware(
    RateLimiterMiddleware,
    rules=rules,
    backend=backend,
)

Полный пример с SlowAPI

from fastapi import FastAPI, Request, Depends
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware

app = FastAPI(title="Rate Limited API")

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["100/minute"],
    storage_uri="redis://localhost:6379"  # или None для in-memory
)

app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)

# Глобальный лимит применяется ко всем эндпоинтам

@app.get("/")
async def root(request: Request):
    return {"message": "Welcome"}

@app.get("/limited")
@limiter.limit("5/minute")
async def limited(request: Request):
    return {"message": "Limited to 5 per minute"}

@app.post("/login")
@limiter.limit("10/hour")
async def login(request: Request, username: str, password: str):
    # Проверка логина
    return {"message": "Login attempt"}

@app.get("/unlimited")
@limiter.exempt
async def unlimited(request: Request):
    return {"message": "No rate limit applied"}

@app.get("/admin")
@limiter.limit("1/minute", key_func=lambda: "admin")
async def admin(request: Request):
    return {"message": "Admin endpoint"}

Задачи для закрепления

Задача 1. Настрой глобальный лимит 100 запросов в минуту.

Задача 2. Добавь per-route лимит 5 запросов в минуту для /login.

Задача 3. Исключи /health из глобального лимита.

Задача 4. Настрой Redis для продакшена.

Задача 5. Добавь кастомный ключ (по user_id вместо IP).

Ответы:

Задача 1.

limiter = Limiter(key_func=get_remote_address, default_limits=["100/minute"])

Задача 2.

@app.post("/login")
@limiter.limit("5/minute")
async def login(request: Request):
    return {"message": "Login"}

Задача 3.

@app.get("/health")
@limiter.exempt
async def health(request: Request):
    return {"status": "ok"}

Задача 4.

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379"
)

Задача 5.

async def get_user_id(request: Request):
    return request.headers.get("X-User-ID", "anonymous")

limiter = Limiter(key_func=get_user_id)

Нюансы и подводные камни

В production используй Redis

In-memory хранилище не работает при нескольких экземплярах приложения .

Исключай health check из лимитов

Health check должны быть доступны всегда, даже при превышении лимитов .

Защита от brute-force

Для эндпоинтов аутентификации используй строгие лимиты (5–10 запросов в минуту) .

Обработка 429

Возвращай понятное сообщение и заголовок Retry-After.

Частые ошибки и как их избежать

Ошибка 1: In-memory в продакшене

Используй Redis для распределённых систем.

Ошибка 2: Одинаковые лимиты для всех эндпоинтов

Для auth эндпоинтов устанавливай более строгие лимиты.

Ошибка 3: Отсутствие исключений для health check

Исключай health check из лимитов.

Ошибка 4: Неправильная идентификация клиента

Используй комбинацию IP + User-Agent для более точной идентификации.

Шпаргалка

Библиотека	Преимущества	Недостатки
SlowAPI	Популярная, гибкая, поддержка Redis	Нужно передавать request в эндпоинт
fastapi-limiter	Dependency-based, простота	Требует Redis
Кастомный	Без зависимостей	Не подходит для продакшена