Middleware в FastAPI

Привет! Middleware — это программное обеспечение, которое обрабатывает запросы и ответы до того, как они достигнут эндпоинта. Это мощный инструмент для логирования, аутентификации, CORS и многого другого.

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

Что такое middleware
Создание кастомного middleware
Встроенные middleware
CORS, логирование, аутентификация
Порядок выполнения

Содержание

Что нужно знать перед началом
Что нужно знать перед началом
Создание кастомного middleware
Встроенные middleware
CORS (Cross-Origin Resource Sharing)
TrustedHostMiddleware
Middleware для аутентификации
Middleware для логирования запросов
Порядок выполнения middleware
Полный пример
Задачи для закрепления
Нюансы и подводные камни
Порядок middleware важен
middleware и исключения
request.state
Производительность
Частые ошибки и как их избежать
Ошибка 1: Забыл await перед call_next
Ошибка 2: Неправильный порядок CORS
Ошибка 3: Изменение response после await call_next
Шпаргалка
Заключение
КВИЗ
Что дальше?

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

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

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

Совет: Middleware — это «прослойка» между клиентом и эндпоинтом. Она может изменять запрос или ответ.

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

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

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

Совет: Middleware — это «прослойка» между клиентом и эндпоинтом. Она может изменять запрос или ответ.

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Код ДО обработки запроса
        print("Middleware: до эндпоинта")

        response = await call_next(request)

        # Код ПОСЛЕ обработки запроса
        print("Middleware: после эндпоинта")
        return response

app.add_middleware(CustomMiddleware)

Совет: call_next(request) передаёт запрос дальше (в эндпоинт или следующий middleware).

Создание кастомного middleware

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

app = FastAPI()

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Запоминаем время начала
        start_time = time.time()

        # Логируем запрос
        print(f"Запрос: {request.method} {request.url.path}")

        # Передаём запрос дальше
        response = await call_next(request)

        # Логируем ответ
        process_time = time.time() - start_time
        print(f"Ответ: {response.status_code} (время: {process_time:.4f} сек)")

        # Добавляем заголовок с временем выполнения
        response.headers["X-Process-Time"] = str(process_time)
        return response

app.add_middleware(LoggingMiddleware)

@app.get("/")
async def root():
    return {"message": "Hello, World!"}

Встроенные middleware

FastAPI предоставляет несколько встроенных middleware из Starlette.

CORS позволяет разрешать запросы с других доменов.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Разрешаем все домены
    allow_credentials=True,
    allow_methods=["*"],  # Все методы
    allow_headers=["*"],  # Все заголовки
)

Совет: В продакшене указывай конкретные домены вместо ["*"].

TrustedHostMiddleware

Защищает от атак с подменой хоста.

from starlette.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

Middleware для аутентификации

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

app = FastAPI()

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Проверяем заголовок Authorization
        auth_header = request.headers.get("Authorization")

        # Пропускаем публичные маршруты
        if request.url.path in ["/login", "/register"]:
            return await call_next(request)

        if not auth_header or not auth_header.startswith("Bearer "):
            return Response("Unauthorized", status_code=401)

        token = auth_header.split(" ")[1]
        # Здесь должна быть проверка токена
        if token != "secret":
            return Response("Invalid token", status_code=401)

        # Сохраняем пользователя в request.state
        request.state.user = {"id": 1, "name": "Анна"}
        return await call_next(request)

app.add_middleware(AuthMiddleware)

@app.get("/protected/")
async def protected(request: Request):
    user = getattr(request.state, "user", None)
    if user:
        return {"message": f"Привет, {user['name']}!"}
    return {"message": "Ошибка"}

Совет: request.state — это специальный объект для хранения данных в рамках одного запроса.

Middleware для логирования запросов

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

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Логируем запрос
        logger.info(f"REQUEST: {request.method} {request.url.path}")

        # Сохраняем время
        start_time = time.time()

        response = await call_next(request)

        # Логируем ответ
        process_time = time.time() - start_time
        logger.info(
            f"RESPONSE: {request.method} {request.url.path} "
            f"-> {response.status_code} ({process_time:.4f} сек)"
        )

        return response

app.add_middleware(RequestLoggingMiddleware)

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

Middleware выполняются в порядке добавления. Первый добавленный — ближе к клиенту.

# Добавляем middleware (порядок важен!)
app.add_middleware(CORSMiddleware)      # Самый внешний
app.add_middleware(LoggingMiddleware)    # Внутри CORS
app.add_middleware(AuthMiddleware)       # Самый внутренний

# Порядок выполнения:
# 1. CORS
# 2. Logging
# 3. Auth
# 4. Эндпоинт
# 5. Auth
# 6. Logging
# 7. CORS

Совет: Порядок важен! Добавляй middleware в правильной последовательности.

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

from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="API with Middleware")

# 1. Логирование запросов
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        logger.info(f"Request: {request.method} {request.url.path}")
        response = await call_next(request)
        process_time = time.time() - start_time
        logger.info(f"Response: {response.status_code} ({process_time:.4f}s)")
        response.headers["X-Process-Time"] = str(process_time)
        return response

# 2. Аутентификация
class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in ["/login", "/", "/docs", "/openapi.json"]:
            return await call_next(request)

        token = request.headers.get("Authorization")
        if not token or not token.startswith("Bearer "):
            return Response("Unauthorized", status_code=401)

        # Проверка токена
        if token != "Bearer secret":
            return Response("Invalid token", status_code=401)

        request.state.user = {"id": 1, "name": "Анна"}
        return await call_next(request)

# 3. Добавляем middleware (порядок важен!)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
app.add_middleware(LoggingMiddleware)
app.add_middleware(AuthMiddleware)

# Эндпоинты
@app.get("/")
async def root():
    return {"message": "Welcome!"}

@app.get("/protected/")
async def protected(request: Request):
    user = getattr(request.state, "user", None)
    return {"message": f"Hello, {user['name'] if user else 'Guest'}"}

@app.get("/public/")
async def public():
    return {"message": "Public endpoint"}

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

Задача 1. Создай middleware для логирования времени выполнения.

Задача 2. Добавь CORS с разрешёнными доменами.

Задача 3. Создай middleware для проверки аутентификации.

Задача 4. Добавь заголовок с версией API в каждый ответ.

Задача 5. Напиши middleware, который добавляет request.start_time.

Ответы:

Задача 1.

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        print(f"Time: {time.time() - start:.4f}s")
        return response

Задача 2.

app.add_middleware(
    CORSMiddleware,
    allow_origins=["example.com"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

Задача 3.

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if "Authorization" not in request.headers:
            return Response("Unauthorized", status_code=401)
        return await call_next(request)

Задача 4.

class VersionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-API-Version"] = "1.0.0"
        return response

Задача 5.

class TimerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request.state.start_time = time.time()
        return await call_next(request)

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

Порядок middleware важен

Middleware выполняются в порядке добавления. Изменение порядка может изменить поведение.

middleware и исключения

Если middleware выбрасывает исключение, оно не будет обработано эндпоинтом.

request.state

Используй request.state для хранения данных между middleware и эндпоинтом.

Производительность

Middleware добавляет накладные расходы. Не добавляй тяжёлую логику без необходимости.

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

Ошибка 1: Забыл `await` перед `call_next`

Неправильно:

response = call_next(request)   # Ошибка! Не дожидаемся

Правильно:

response = await call_next(request)

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

CORS должен быть первым.

Ошибка 3: Изменение response после `await call_next`

Изменяй response после получения.

Шпаргалка

Компонент	Назначение
`BaseHTTPMiddleware`	Базовый класс для middleware
`CORSMiddleware`	Настройка CORS
`TrustedHostMiddleware`	Защита от подмены хоста
`request.state`	Хранение данных между middleware и эндпоинтом
`await call_next(request)`	Передача запроса дальше

Заключение

Сегодня мы:

Узнали, что такое middleware
Создали кастомные middleware
Настроили CORS и аутентификацию
Разобрали порядок выполнения

КВИЗ

Что дальше?

Следующая статья: «CORS в FastAPI»

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

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

Создание кастомного middleware

Встроенные middleware

CORS (Cross-Origin Resource Sharing)

TrustedHostMiddleware

Middleware для аутентификации

Middleware для логирования запросов

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

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

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

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

Порядок middleware важен

middleware и исключения

request.state

Производительность

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

Ошибка 1: Забыл await перед call_next

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

Ошибка 3: Изменение response после await call_next

Шпаргалка

Заключение

КВИЗ

Что дальше?

Ошибка 1: Забыл `await` перед `call_next`

Ошибка 3: Изменение response после `await call_next`