Middleware — FastAPI Pro: От новичка до архитектора

Middleware в FastAPI

Middleware обрабатывает запросы до и после endpoints. В этой теме вы научитесь логированию, CORS, rate limiting и созданию кастомных middleware.

#Что такое Middleware?

Middleware — функция, выполняющаяся до и после обработки запроса endpoint'ом.

Запрос → Middleware 1 → Middleware 2 → Endpoint → Response → Middleware 2 → Middleware 1 → Клиент

#Когда использовать?

Задача	Middleware	Почему
Логирование запросов	✅	Все запросы проходят через middleware
CORS	✅	Добавляет заголовки ко всем ответам
Rate limiting	✅	Проверка до выполнения endpoint
Аутентификация	✅	Проверка токена до доступа
Сжатие ответов	✅	Обработка всех ответов
Валидация данных	❌	Делается в endpoint через Pydantic

#Базовый middleware

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """
    Логирование всех запросов.
    """
    # До выполнения
    start_time = time.time()
    
    print(f"{request.method} {request.url.path}")
    
    # Выполнение запроса
    response = await call_next(request)
    
    # После выполнения
    duration = time.time() - start_time
    
    print(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s")
    
    return response

#Middleware с классом

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class CustomHeaderMiddleware(BaseHTTPMiddleware):
    """
    Добавляет кастомные заголовки ко всем ответам.
    """
    
    async def dispatch(self, request: Request, call_next):
        # До выполнения
        request.state.custom_data = "some data"
        
        # Выполнение
        response = await call_next(request)
        
        # После выполнения
        response.headers['X-Custom-Header'] = 'Custom Value'
        response.headers['X-Request-ID'] = request.headers.get('X-Request-ID', 'unknown')
        
        return response

app.add_middleware(CustomHeaderMiddleware)

#CORS Middleware

CORS (Cross-Origin Resource Sharing) — разрешение запросов с других доменов.

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Настройка CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",      # React
        "http://localhost:8080",      # Vue
        "https://myapp.com",          # Production
    ],
    allow_credentials=True,           # Разрешить cookies
    allow_methods=["*"],              # Все методы (GET, POST, PUT, DELETE)
    allow_headers=["*"],              # Все заголовки
)

#Параметры CORS

Параметр	Значение	Описание
`allow_origins`	`["https://site.com"]`	Разрешённые домены
`allow_credentials`	`True/False`	Разрешить cookies/authorization
`allow_methods`	`["GET", "POST"]`	Разрешённые методы
`allow_headers`	`["*"]`	Разрешённые заголовки

#Все разрешения (только для разработки!)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],      # Все домены
    allow_credentials=False,  # Cookies не работают с "*"
    allow_methods=["*"],
    allow_headers=["*"],
)

#Rate Limiting Middleware

Ограничение количества запросов от одного клиента.

#Простой rate limiter

from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import JSONResponse
from collections import defaultdict
import time

app = FastAPI()

# Хранилище: {ip: [(timestamp, count)]}
request_history = defaultdict(list)

RATE_LIMIT = 10  # запросов
TIME_WINDOW = 60  # секунд

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    current_time = time.time()
    
    # Очистка старых записей
    request_history[client_ip] = [
        (ts, count) for ts, count in request_history[client_ip]
        if current_time - ts < TIME_WINDOW
    ]
    
    # Подсчёт запросов
    total_requests = sum(count for _, count in request_history[client_ip])
    
    if total_requests >= RATE_LIMIT:
        return JSONResponse(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            content={"detail": "Rate limit exceeded. Try again later."}
        )
    
    # Добавление текущего запроса
    request_history[client_ip].append((current_time, 1))
    
    # Выполнение запроса
    response = await call_next(request)
    
    # Добавление заголовков
    response.headers['X-RateLimit-Limit'] = str(RATE_LIMIT)
    response.headers['X-RateLimit-Remaining'] = str(RATE_LIMIT - total_requests - 1)
    
    return response

#Rate limiter с Redis

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import redis
import time

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379)

RATE_LIMIT = 100  # запросов в минуту
TIME_WINDOW = 60  # секунд

@app.middleware("http")
async def redis_rate_limit(request: Request, call_next):
    client_ip = request.client.host
    key = f"rate_limit:{client_ip}"
    
    # Получение текущего счёта
    current = redis_client.get(key)
    
    if current and int(current) >= RATE_LIMIT:
        ttl = redis_client.ttl(key)
        
        return JSONResponse(
            status_code=429,
            content={
                "detail": "Rate limit exceeded",
                "retry_after": ttl
            },
            headers={"Retry-After": str(ttl)}
        )
    
    # Инкремент счёта
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, TIME_WINDOW)
    pipe.execute()
    
    # Выполнение запроса
    response = await call_next(request)
    
    # Заголовки
    remaining = RATE_LIMIT - int(redis_client.get(key) or 0)
    response.headers['X-RateLimit-Limit'] = str(RATE_LIMIT)
    response.headers['X-RateLimit-Remaining'] = str(max(0, remaining))
    
    return response

#GZip Middleware

Сжатие ответов для уменьшения трафика.

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)  # Сжимать от 1KB

#Настройка

Параметр	Значение	Описание
`minimum_size`	`1000`	Минимальный размер для сжатия (байты)
`compress_level`	`5`	Уровень сжатия (1-9, по умолчанию 5)

#HTTPS Redirect Middleware

Перенаправление с HTTP на HTTPS.

from fastapi import FastAPI, Request
from starlette.responses import RedirectResponse

app = FastAPI()

@app.middleware("http")
async def https_redirect(request: Request, call_next):
    # Только для production
    if request.url.scheme == "http" and request.url.hostname != "localhost":
        url = request.url.replace(scheme="https")
        return RedirectResponse(url, status_code=301)
    
    return await call_next(request)

#Security Headers Middleware

Добавление заголовков безопасности.

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def security_headers(request: Request, call_next):
    response = await call_next(request)
    
    # Защита от XSS
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
    
    # Content Security Policy
    response.headers['Content-Security-Policy'] = (
        "default-src 'self'; "
        "script-src 'self'; "
        "style-src 'self' 'unsafe-inline'"
    )
    
    return response

#Timing Middleware

Добавление заголовка с временем выполнения.

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    response.headers['X-Process-Time'] = str(duration)
    
    return response

#Пример: Полный набор middleware

from fastapi import FastAPI, Request, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging
import redis

app = FastAPI()

# Логирование
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Redis для rate limiting
redis_client = redis.Redis(host='localhost', port=6379)

# === Middleware ===

# 1. CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 2. GZip
app.add_middleware(GZipMiddleware, minimum_size=1000)

# 3. Логирование
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    
    logger.info(f"{request.method} {request.url.path}")
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    logger.info(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s")
    
    response.headers['X-Process-Time'] = str(duration)
    
    return response

# 4. Rate Limiting
RATE_LIMIT = 100
TIME_WINDOW = 60

@app.middleware("http")
async def rate_limit(request: Request, call_next):
    client_ip = request.client.host
    key = f"rate:{client_ip}"
    
    current = redis_client.get(key)
    
    if current and int(current) >= RATE_LIMIT:
        raise HTTPException(
            status_code=429,
            detail="Too many requests",
            headers={"Retry-After": str(redis_client.ttl(key))}
        )
    
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, TIME_WINDOW)
    pipe.execute()
    
    response = await call_next(request)
    
    remaining = RATE_LIMIT - int(redis_client.get(key) or 0)
    response.headers['X-RateLimit-Remaining'] = str(remaining)
    
    return response

# 5. Security Headers
@app.middleware("http")
async def security_headers(request: Request, call_next):
    response = await call_next(request)
    
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    
    return response

# === Endpoints ===

@app.get('/')
def read_root():
    return {'message': 'Hello World'}

@app.get('/slow')
def slow_endpoint():
    time.sleep(2)
    return {'message': 'Slow response'}

@app.get('/large')
def large_endpoint():
    # Большой ответ (будет сжат GZip)
    return {'data': 'x' * 10000}

#Порядок middleware

Middleware выполняются в порядке добавления:

app.add_middleware(Middleware1)  # Выполняется первым ДО, последним ПОСЛЕ
app.add_middleware(Middleware2)  # Выполняется вторым ДО, первым ПОСЛЕ
app.add_middleware(Middleware3)  # Выполняется последним ДО, вторым ПОСЛЕ

@app.middleware("http")
def middleware4():  # Выполняется после всех add_middleware ДО, первым ПОСЛЕ
    ...

#Тестирование middleware

from fastapi.testclient import TestClient

client = TestClient(app)

def test_rate_limit_headers():
    response = client.get('/')
    
    assert 'X-RateLimit-Remaining' in response.headers
    assert 'X-Process-Time' in response.headers
    assert 'X-Content-Type-Options' in response.headers

def test_cors_headers():
    response = client.get('/', headers={'Origin': 'https://myapp.com'})
    
    assert 'access-control-allow-origin' in response.headers

def test_gzip():
    response = client.get('/large', headers={'Accept-Encoding': 'gzip'})
    
    assert response.headers.get('Content-Encoding') == 'gzip'

#Частые ошибки

#Ошибка 1: Блокирующий код в middleware

@app.middleware("http")
async def bad_middleware(request: Request, call_next):
    time.sleep(1)  # Блокирует все запросы!
    return await call_next(request)

Решение: Используйте await asyncio.sleep(1).

#Ошибка 2: Забыли вызвать call_next

@app.middleware("http")
async def bad_middleware(request: Request, call_next):
    # Забыли call_next!
    return Response("Always same response")

Решение: Всегда вызывайте response = await call_next(request).

#Ошибка 3: Неправильный порядок CORS

@app.middleware("http")
def custom_middleware(request, call_next):
    response = call_next(request)
    response.headers['Access-Control-Allow-Origin'] = '*'  # Не работает!
    return response

Решение: Используйте CORSMiddleware от FastAPI, он обрабатывает preflight запросы.

#Исполняемая лаборатория

Лаборатория «Тайминг, correlation ID, обработка ошибок» доступна в публичном GitLab. Она проверяет кастомный middleware для тайминга запросов, генерацию и прокидывание correlation ID, обработку ошибок в middleware и правильный порядок регистрации middleware.

Склонируйте репозиторий и запустите тесты из его корня:

git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest middleware/tests

Задача

Middleware

Почему

Логирование запросов

✅

Все запросы проходят через middleware

CORS

✅

Добавляет заголовки ко всем ответам

Rate limiting

✅

Проверка до выполнения endpoint

Аутентификация

✅

Проверка токена до доступа

Сжатие ответов

✅

Обработка всех ответов

Валидация данных

❌

Делается в endpoint через Pydantic

Параметр

Значение

Описание

allow_origins

["https://site.com"]

Разрешённые домены

allow_credentials

True/False

Разрешить cookies/authorization

allow_methods

["GET", "POST"]

Разрешённые методы

allow_headers

["*"]

Разрешённые заголовки

Параметр

Значение

Описание

minimum_size

1000

Минимальный размер для сжатия (байты)

compress_level

5

Уровень сжатия (1-9, по умолчанию 5)