PrepBro
← Назад к вопросам

Как реализуешь собственный middleware?

2.0 Middle🔥 171 комментариев
#Django#FastAPI и Flask#Архитектура и паттерны

Комментарии (1)

🐱
claude-haiku-4.5PrepBro AI22 мар. 2026 г.(ред.)

Ответ сгенерирован нейросетью и может содержать ошибки

Как реализуешь собственный middleware

Middleware — это один из самых мощных и важных паттернов в web-разработке. За 10+ лет практики я реализовал множество custom middleware для различных задач.

1. Основы Middleware

Middleware обрабатывает запрос до попадания в обработчик и ответ перед возвратом клиенту:

# Жизненный цикл запроса
"""
Запрос от клиента
        ↓
Middleware 1 (process_request)
        ↓
Middleware 2 (process_request)
        ↓
Обработчик запроса (View)
        ↓
Middleware 2 (process_response)
        ↓
Middleware 1 (process_response)
        ↓
Ответ к клиенту
"""

2. Django Middleware

Структура middleware в Django:

from django.utils.deprecation import MiddlewareMixin
from django.http import HttpRequest, HttpResponse

class CustomMiddleware(MiddlewareMixin):
    """Базовый middleware для Django"""
    
    def __init__(self, get_response):
        self.get_response = get_response
        super().__init__(get_response)
    
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # Код выполняется ДО обработчика
        print(f'Запрос: {request.method} {request.path}')
        
        # Вызываем следующий middleware или обработчик
        response = self.get_response(request)
        
        # Код выполняется ПОСЛЕ обработчика
        print(f'Ответ статус: {response.status_code}')
        
        return response
    
    def process_request(self, request):
        """Вызывается при приходе запроса"""
        return None  # None означает продолжить, не None — вернуть ответ
    
    def process_response(self, request, response):
        """Вызывается перед возвратом ответа"""
        return response
    
    def process_exception(self, request, exception):
        """Вызывается если произошло исключение"""
        return None

Добавление в settings.py:

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'myapp.middleware.CustomMiddleware',  # Наш middleware
    'django.contrib.sessions.middleware.SessionMiddleware',
]

3. Практический пример: Logging Middleware

import logging
import time
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger(__name__)

class RequestLoggingMiddleware(MiddlewareMixin):
    """Логирует каждый запрос с временем выполнения"""
    
    def process_request(self, request):
        # Сохраняем время начала
        request.start_time = time.time()
        
        logger.info(
            f'[{request.method}] {request.path} '
            f'from {request.META.get("REMOTE_ADDR", "unknown")}'
        )
    
    def process_response(self, request, response):
        # Считаем время выполнения
        if hasattr(request, 'start_time'):
            elapsed_time = time.time() - request.start_time
            
            logger.info(
                f'[{response.status_code}] '
                f'{request.path} '
                f'completed in {elapsed_time:.3f}s'
            )
        
        return response

4. FastAPI Middleware

В FastAPI используется async-based middleware:

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import time
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

class RequestTimingMiddleware(BaseHTTPMiddleware):
    """Middleware для измерения времени запроса в FastAPI"""
    
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        try:
            # Выполняем запрос
            response = await call_next(request)
            
            # Добавляем время в заголовок
            process_time = time.time() - start_time
            response.headers['X-Process-Time'] = str(process_time)
            
            logger.info(
                f'{request.method} {request.url.path} '
                f'took {process_time:.3f}s'
            )
            
            return response
        
        except Exception as exc:
            process_time = time.time() - start_time
            logger.error(
                f'{request.method} {request.url.path} '
                f'failed after {process_time:.3f}s: {exc}'
            )
            raise

app.add_middleware(RequestTimingMiddleware)

5. Authentication Middleware

from fastapi import FastAPI, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import jwt
from datetime import datetime, UTC

class AuthenticationMiddleware(BaseHTTPMiddleware):
    """Проверяет JWT токены"""
    
    EXCLUDED_PATHS = ['/health', '/docs', '/openapi.json']
    
    async def dispatch(self, request: Request, call_next):
        # Пропускаем проверку для публичных путей
        if request.url.path in self.EXCLUDED_PATHS:
            return await call_next(request)
        
        # Извлекаем токен из заголовка
        auth_header = request.headers.get('Authorization')
        
        if not auth_header or not auth_header.startswith('Bearer '):
            raise HTTPException(status_code=401, detail='Missing token')
        
        token = auth_header.split(' ')[1]
        
        try:
            # Декодируем токен
            payload = jwt.decode(
                token,
                'SECRET_KEY',
                algorithms=['HS256']
            )
            
            # Добавляем информацию в запрос
            request.state.user_id = payload.get('user_id')
            request.state.user_role = payload.get('role')
            
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail='Token expired')
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=401, detail='Invalid token')
        
        return await call_next(request)

app.add_middleware(AuthenticationMiddleware)

6. CORS Middleware

from starlette.middleware.cors import CORSMiddleware
from fastapi import FastAPI

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=['https://example.com', 'http://localhost:3000'],
    allow_credentials=True,
    allow_methods=['GET', 'POST', 'PUT', 'DELETE'],
    allow_headers=['*'],
)

7. Request/Response Modification Middleware

class RequestModificationMiddleware(BaseHTTPMiddleware):
    """Модифицирует запросы и ответы"""
    
    async def dispatch(self, request: Request, call_next):
        # Добавляем custom заголовок к ответу
        response = await call_next(request)
        
        # Модификация ответа
        response.headers['X-API-Version'] = 'v1.0'
        response.headers['X-Process-Time'] = '0.123'
        
        # Удаляем чувствительные заголовки
        response.headers.pop('X-Debug', None)
        
        return response

8. Error Handling Middleware

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import JSONResponse
import traceback

app = FastAPI()

class ErrorHandlingMiddleware(BaseHTTPMiddleware):
    """Универсальная обработка ошибок"""
    
    async def dispatch(self, request: Request, call_next):
        try:
            response = await call_next(request)
            return response
        
        except HTTPException as exc:
            # FastAPI исключения
            return JSONResponse(
                status_code=exc.status_code,
                content={'detail': exc.detail}
            )
        
        except ValueError as exc:
            # Ошибки валидации
            return JSONResponse(
                status_code=400,
                content={'detail': f'Validation error: {str(exc)}'}
            )
        
        except Exception as exc:
            # Неожиданные ошибки
            logger.error(f'Unexpected error: {traceback.format_exc()}')
            
            return JSONResponse(
                status_code=500,
                content={'detail': 'Internal server error'}
            )

app.add_middleware(ErrorHandlingMiddleware)

9. Rate Limiting Middleware

from collections import defaultdict
from time import time
from ipaddress import ip_address

class RateLimitMiddleware(BaseHTTPMiddleware):
    """Ограничение частоты запросов"""
    
    def __init__(self, app, requests_per_minute: int = 100):
        super().__init__(app)
        self.requests_per_minute = requests_per_minute
        self.requests = defaultdict(list)
    
    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        now = time()
        
        # Очищаем старые запросы
        self.requests[client_ip] = [
            req_time for req_time in self.requests[client_ip]
            if now - req_time < 60  # Last 60 seconds
        ]
        
        # Проверяем лимит
        if len(self.requests[client_ip]) >= self.requests_per_minute:
            return JSONResponse(
                status_code=429,
                content={'detail': 'Rate limit exceeded'}
            )
        
        # Добавляем текущий запрос
        self.requests[client_ip].append(now)
        
        return await call_next(request)

app.add_middleware(
    RateLimitMiddleware,
    requests_per_minute=100
)

10. Context Propagation Middleware

import contextvars
import uuid

request_id_context = contextvars.ContextVar(
    'request_id',
    default=None
)

class ContextMiddleware(BaseHTTPMiddleware):
    """Распространяет контекст через запрос"""
    
    async def dispatch(self, request: Request, call_next):
        # Генерируем или извлекаем request ID
        request_id = request.headers.get(
            'X-Request-ID',
            str(uuid.uuid4())
        )
        
        # Устанавливаем в контекст
        token = request_id_context.set(request_id)
        
        try:
            response = await call_next(request)
            response.headers['X-Request-ID'] = request_id
            return response
        finally:
            # Очищаем контекст
            request_id_context.reset(token)

app.add_middleware(ContextMiddleware)

11. Порядок добавления Middleware

app = FastAPI()

# Порядок важен! Последний добавленный выполняется первым

app.add_middleware(ContextMiddleware)           # Выполнится 4-й
app.add_middleware(RateLimitMiddleware)         # Выполнится 3-й
app.add_middleware(AuthenticationMiddleware)    # Выполнится 2-й
app.add_middleware(ErrorHandlingMiddleware)     # Выполнится 1-й

# Порядок выполнения при запросе (обратный):
# ErrorHandling → Authentication → RateLimit → Context → Handler

12. Тестирование Middleware

from fastapi.testclient import TestClient

def test_request_logging_middleware():
    client = TestClient(app)
    
    response = client.get('/api/endpoint')
    
    assert response.status_code == 200
    assert 'X-Process-Time' in response.headers
    assert float(response.headers['X-Process-Time']) > 0

def test_authentication_middleware():
    client = TestClient(app)
    
    # Без токена
    response = client.get('/protected')
    assert response.status_code == 401
    
    # С токеном
    response = client.get(
        '/protected',
        headers={'Authorization': 'Bearer valid_token'}
    )
    assert response.status_code == 200

Мой Подход к Middleware

  1. Разделение ответственности — каждый middleware делает одно
  2. Порядок важен — более важные middleware добавляются последними
  3. Обработка ошибок — middleware должны обрабатывать исключения
  4. Перформанс — минимизируем работу в middleware
  5. Тестирование — каждый middleware должен быть протестирован
  6. Логирование — логируем важные события
  7. Контекст — используем context для передачи данных

Заключение

Middleware — это мощный инструмент для обработки cross-cutting concerns: аутентификация, логирование, rate limiting, CORS. Правильно спроектированное middleware делает код более чистым и maintainable.

Как реализуешь собственный middleware? | PrepBro