Промежуточное ПО (Middleware)

В этом разделе описывается промежуточное ПО (middleware), используемое в Boilerplate. Вы узнаете, как middleware обрабатывает запросы и ответы, а также как создавать собственные middleware компоненты.

Обзор Middleware

Middleware в FastAPI — это компоненты, которые обрабатывают HTTP-запросы и ответы до и после их обработки основными обработчиками. Middleware может использоваться для:

Логирования запросов и ответов
Аутентификации и авторизации
Обработки CORS (Cross-Origin Resource Sharing)
Мониторинга производительности и метрик
Модификации заголовков запросов и ответов
Обработки исключений

Стандартные middleware в Boilerplate

В Boilerplate используются следующие стандартные middleware компоненты:

CORS Middleware

CORS (Cross-Origin Resource Sharing) middleware позволяет контролировать доступ к API с других доменов:

python

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.core.config import settings

app = FastAPI()

# Настройка CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.BACKEND_CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Настройки CORS в app/core/config.py:

python

# app/core/config.py
from typing import List
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # ... другие настройки ...
    
    # CORS
    BACKEND_CORS_ORIGINS: List[str] = ["http://localhost:3000", "https://example.com"]
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        case_sensitive = True

settings = Settings()

GZip Middleware

GZip middleware сжимает ответы для уменьшения объема передаваемых данных:

python

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# Добавляем GZip сжатие
app.add_middleware(GZipMiddleware, minimum_size=1000)

Trusted Host Middleware

Middleware для проверки доверенных хостов, защищающее от атак с подменой хоста:

python

# app/main.py
from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddleware

from app.core.config import settings

app = FastAPI()

# Добавляем проверку доверенных хостов в production
if not settings.DEBUG:
    app.add_middleware(
        TrustedHostMiddleware, allowed_hosts=settings.ALLOWED_HOSTS
    )

Кастомные Middleware

В Boilerplate реализованы следующие пользовательские middleware компоненты:

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

python

# app/middleware/logging.py
import time
from typing import Callable

from fastapi import FastAPI, Request, Response
from loguru import logger
from starlette.middleware.base import BaseHTTPMiddleware

class RequestLoggingMiddleware(BaseHTTPMiddleware):
    """
    Middleware для логирования HTTP запросов и ответов.
    """
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        # Записываем время начала запроса
        start_time = time.time()
        
        # Формируем информацию о запросе
        request_id = request.headers.get("X-Request-ID", "")
        request_info = {
            "request_id": request_id,
            "method": request.method,
            "url": str(request.url),
            "client": request.client.host if request.client else "",
            "user_agent": request.headers.get("User-Agent", ""),
        }
        
        # Логируем запрос
        logger.info(f"Request: {request.method} {request.url.path}", request_info)
        
        # Обрабатываем запрос
        try:
            response = await call_next(request)
            
            # Вычисляем время обработки
            process_time = time.time() - start_time
            
            # Добавляем заголовок с временем обработки
            response.headers["X-Process-Time"] = str(process_time)
            
            # Логируем ответ
            response_info = {
                **request_info,
                "status_code": response.status_code,
                "process_time": process_time,
            }
            logger.info(
                f"Response: {response.status_code} {request.method} {request.url.path}",
                response_info
            )
            
            return response
        except Exception as e:
            # Логируем ошибку
            error_info = {
                **request_info,
                "error": str(e),
                "error_type": type(e).__name__,
            }
            logger.error(
                f"Error during request processing: {request.method} {request.url.path}",
                error_info
            )
            raise

def add_logging_middleware(app: FastAPI) -> None:
    """
    Добавляет middleware для логирования запросов.
    """
    app.add_middleware(RequestLoggingMiddleware)

Middleware для аутентификации с API-ключом

python

# app/middleware/api_key.py
from typing import Callable, List, Optional

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

from app.core.config import settings

class APIKeyMiddleware(BaseHTTPMiddleware):
    """
    Middleware для аутентификации с использованием API-ключа.
    """
    def __init__(
        self, 
        app: FastAPI, 
        api_key: str,
        exclude_paths: List[str] = None
    ):
        super().__init__(app)
        self.api_key = api_key
        self.exclude_paths = exclude_paths or []
    
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        # Проверяем, нужно ли исключить путь из проверки
        path = request.url.path
        if any(path.startswith(exclude_path) for exclude_path in self.exclude_paths):
            return await call_next(request)
        
        # Получаем API-ключ из заголовка
        api_key = request.headers.get("X-API-Key")
        
        # Проверяем API-ключ
        if not api_key or api_key != self.api_key:
            return JSONResponse(
                status_code=401,
                content={
                    "success": False,
                    "error": {
                        "code": 401,
                        "message": "Неверный API-ключ",
                        "type": "unauthorized"
                    }
                }
            )
        
        # Если ключ верный, продолжаем обработку
        return await call_next(request)

def add_api_key_middleware(
    app: FastAPI, 
    exclude_paths: Optional[List[str]] = None
) -> None:
    """
    Добавляет middleware для аутентификации с API-ключом.
    """
    if settings.API_KEY_ENABLED:
        app.add_middleware(
            APIKeyMiddleware,
            api_key=settings.API_KEY,
            exclude_paths=exclude_paths or ["/docs", "/redoc", "/openapi.json"]
        )

Middleware для управления транзакциями

python

# app/middleware/transaction.py
from typing import Callable

from fastapi import FastAPI, Request, Response
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.middleware.base import BaseHTTPMiddleware

from app.db.session import AsyncSessionLocal

class TransactionMiddleware(BaseHTTPMiddleware):
    """
    Middleware для управления транзакциями базы данных.
    Автоматически создает сессию для каждого запроса и
    выполняет commit или rollback в зависимости от результата.
    """
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        # Создаем сессию
        async with AsyncSessionLocal() as session:
            # Добавляем сессию в состояние запроса
            request.state.db = session
            
            try:
                # Обрабатываем запрос
                response = await call_next(request)
                
                # Если статус успешный, коммитим транзакцию
                if 200 <= response.status_code < 400:
                    await session.commit()
                else:
                    # Иначе откатываем
                    await session.rollback()
                
                return response
            except Exception as e:
                # В случае ошибки откатываем транзакцию
                await session.rollback()
                raise

def add_transaction_middleware(app: FastAPI) -> None:
    """
    Добавляет middleware для управления транзакциями.
    """
    app.add_middleware(TransactionMiddleware)

Регистрация Middleware

В файле app/main.py регистрируются все middleware компоненты:

python

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware

from app.core.config import settings
from app.middleware.logging import add_logging_middleware
from app.middleware.api_key import add_api_key_middleware
from app.middleware.transaction import add_transaction_middleware

app = FastAPI(
    title="Boilerplate API",
    description="API для Boilerplate приложения",
    version="0.1.0",
)

# Добавляем стандартные middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.BACKEND_CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=1000)

if not settings.DEBUG:
    app.add_middleware(
        TrustedHostMiddleware, allowed_hosts=settings.ALLOWED_HOSTS
    )

# Добавляем кастомные middleware
add_logging_middleware(app)
add_api_key_middleware(app)
add_transaction_middleware(app)

# ... остальная часть main.py

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

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

text

Порядок выполнения при запросе:
1. TransactionMiddleware (последний зарегистрированный)
2. APIKeyMiddleware
3. RequestLoggingMiddleware
4. TrustedHostMiddleware (если включен)
5. GZipMiddleware
6. CORSMiddleware (первый зарегистрированный)
7. Обработчик маршрута FastAPI

Порядок выполнения при ответе:
1. Обработчик маршрута FastAPI
2. CORSMiddleware
3. GZipMiddleware
4. TrustedHostMiddleware (если включен)
5. RequestLoggingMiddleware
6. APIKeyMiddleware
7. TransactionMiddleware

Учитывайте этот порядок при разработке и добавлении новых middleware, поскольку это может влиять на поведение приложения.

Создание собственного Middleware

В FastAPI есть два способа создания middleware:

Функциональный способ

python

# app/middleware/simple.py
from fastapi import FastAPI, Request

async def simple_middleware(request: Request, call_next):
    """
    Простой функциональный middleware.
    """
    # Действия до обработки запроса
    print(f"Запрос: {request.method} {request.url.path}")
    
    # Обработка запроса
    response = await call_next(request)
    
    # Действия после обработки запроса
    print(f"Ответ: {response.status_code}")
    
    return response

def add_simple_middleware(app: FastAPI) -> None:
    """
    Добавляет простой middleware.
    """
    app.middleware("http")(simple_middleware)

Классовый способ (на базе BaseHTTPMiddleware)

python

# app/middleware/class_based.py
from typing import Callable

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

class ClassBasedMiddleware(BaseHTTPMiddleware):
    """
    Middleware на основе класса.
    """
    def __init__(self, app: FastAPI, some_param: str = "default"):
        super().__init__(app)
        self.some_param = some_param
    
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        # Действия до обработки запроса
        print(f"Параметр: {self.some_param}")
        print(f"Запрос: {request.method} {request.url.path}")
        
        # Обработка запроса
        response = await call_next(request)
        
        # Действия после обработки запроса
        print(f"Ответ: {response.status_code}")
        
        # Модификация ответа (например, добавление заголовка)
        response.headers["X-Custom-Header"] = "Value"
        
        return response

def add_class_based_middleware(app: FastAPI, param: str) -> None:
    """
    Добавляет middleware на основе класса.
    """
    app.add_middleware(ClassBasedMiddleware, some_param=param)

Доступ к объекту сессии базы данных

При использовании TransactionMiddleware объект сессии доступен через request.state.db:

python

# app/api/routes/example.py
from fastapi import APIRouter, Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()

# Функция-зависимость для получения сессии из request.state
async def get_db_from_request(request: Request) -> AsyncSession:
    return request.state.db

@router.get("/example")
async def example_route(db: AsyncSession = Depends(get_db_from_request)):
    """
    Пример маршрута, получающего сессию из middleware.
    """
    # Использование сессии для запросов
    result = await db.execute("SELECT 1")
    
    return {"success": True, "data": result.scalar()}

Middleware для обработки исключений

Middleware для глобальной обработки исключений, в дополнение к обработчикам исключений FastAPI:

python

# app/middleware/exception.py
import traceback
from typing import Callable

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from loguru import logger
from starlette.middleware.base import BaseHTTPMiddleware

from app.core.config import settings

class ExceptionMiddleware(BaseHTTPMiddleware):
    """
    Middleware для глобальной обработки исключений.
    """
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        try:
            # Обрабатываем запрос
            return await call_next(request)
        except Exception as e:
            # Логируем ошибку
            error_info = {
                "method": request.method,
                "url": str(request.url),
                "client": request.client.host if request.client else "",
                "error": str(e),
                "error_type": type(e).__name__,
            }
            
            # В режиме отладки логируем полный стек-трейс
            if settings.DEBUG:
                error_info["traceback"] = traceback.format_exc()
            
            logger.error(f"Unhandled exception: {str(e)}", error_info)
            
            # Формируем ответ с ошибкой
            error_message = str(e) if settings.DEBUG else "Внутренняя ошибка сервера"
            
            return JSONResponse(
                status_code=500,
                content={
                    "success": False,
                    "error": {
                        "code": 500,
                        "message": error_message,
                        "type": "server_error"
                    }
                }
            )

def add_exception_middleware(app: FastAPI) -> None:
    """
    Добавляет middleware для обработки исключений.
    """
    app.add_middleware(ExceptionMiddleware)

Советы и лучшие практики

Размещайте middleware в правильном порядке, учитывая их взаимодействие
Используйте BaseHTTPMiddleware для сложных middleware с дополнительными параметрами
Минимизируйте количество middleware для уменьшения накладных расходов
Не выполняйте тяжелые операции в middleware, если это не абсолютно необходимо
Правильно обрабатывайте исключения в middleware, чтобы избежать "проглатывания" ошибок
Используйте логирование для отладки и мониторинга работы middleware
Создавайте тесты для middleware, чтобы убедиться в их корректной работе

Промежуточное ПО (Middleware)

Обзор Middleware

Стандартные middleware в Boilerplate

CORS Middleware

GZip Middleware

Trusted Host Middleware

Кастомные Middleware

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

Middleware для аутентификации с API-ключом

Middleware для управления транзакциями

Регистрация Middleware

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

Создание собственного Middleware

Функциональный способ

Классовый способ (на базе BaseHTTPMiddleware)

Доступ к объекту сессии базы данных

Middleware для обработки исключений

Советы и лучшие практики

Ресурсы