Приведи пример использования middleware

Question

claude-haiku-4.5 · Accepted Answer

## Middleware в Python: Примеры и Использование

Middleware — это промежуточный слой в приложении, который обрабатывает request и response. Давайте разберём разные примеры в контексте Django и других фреймворков.

## 1. Django Middleware

### Что такое Django Middleware?

Middleware в Django — это класс, который имеет методы для обработки request/response на разных этапах:

```python
# Структура middleware
class MyMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # Код ПЕРЕД обработкой request
        
        response = self.get_response(request)  # Вызов следующего middleware или view
        
        # Код ПОСЛЕ обработки request
        
        return response
```

### Пример 1: Логирование всех запросов

```python
# middleware.py
import logging
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger(__name__)

class RequestLoggingMiddleware(MiddlewareMixin):
    """Middleware для логирования всех HTTP запросов"""
    
    def process_request(self, request):
        # Логируем информацию о request
        logger.info(f"{request.method} {request.path}")
        logger.info(f"GET params: {dict(request.GET)}")
        logger.info(f"POST params: {dict(request.POST)}")
        logger.info(f"User: {request.user}")
    
    def process_response(self, request, response):
        # Логируем status code
        logger.info(f"Response status: {response.status_code}")
        return response
```

### Пример 2: Добавление заголовков безопасности

```python
class SecurityHeadersMiddleware(MiddlewareMixin):
    """Добавляет заголовки безопасности к каждому ответу"""
    
    def process_response(self, request, response):
        # Защита от XSS
        response['X-Content-Type-Options'] = 'nosniff'
        response['X-Frame-Options'] = 'DENY'
        response['X-XSS-Protection'] = '1; mode=block'
        
        # HTTPS only
        response['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
        
        # CSP (Content Security Policy)
        response['Content-Security-Policy'] = "default-src 'self'; script-src 'self' 'unsafe-inline'"
        
        return response
```

### Пример 3: Аутентификация и авторизация

```python
from django.http import HttpResponseForbidden
from django.conf import settings

class AuthenticationMiddleware(MiddlewareMixin):
    """Проверяет, авторизован ли пользователь для определённых URL"""
    
    # Публичные URL, которые не требуют аутентификации
    PUBLIC_URLS = ['/login/', '/register/', '/api/public/']
    
    def process_request(self, request):
        # Если URL публичный, пропускаем проверку
        if any(request.path.startswith(url) for url in self.PUBLIC_URLS):
            return None
        
        # Если пользователь не авторизован
        if not request.user.is_authenticated:
            from django.shortcuts import redirect
            return redirect('login')
        
        return None
```

### Пример 4: Кэширование ответов

```python
from django.core.cache import cache
import hashlib

class CachingMiddleware(MiddlewareMixin):
    """Кэширует GET запросы"""
    
    def process_request(self, request):
        # Кэшируем только GET запросы
        if request.method != 'GET':
            return None
        
        # Создаём уникальный ключ для кэша
        cache_key = f"page:{request.path}:{request.GET.urlencode()}"
        
        # Пытаемся получить из кэша
        cached_response = cache.get(cache_key)
        if cached_response:
            return cached_response
        
        # Сохраняем ключ в request для использования в response middleware
        request._cache_key = cache_key
        return None
    
    def process_response(self, request, response):
        # Если в request есть cache_key, кэшируем ответ
        if hasattr(request, '_cache_key') and response.status_code == 200:
            cache.set(request._cache_key, response, timeout=3600)  # На 1 час
        
        return response
```

### Пример 5: Обработка ошибок и исключений

```python
import traceback
from django.http import JsonResponse
from django.views.debug import get_exception_reporter_filter

class ErrorHandlingMiddleware(MiddlewareMixin):
    """Обрабатывает необработанные исключения"""
    
    def process_exception(self, request, exception):
        # Логируем ошибку
        logger.error(f"Unhandled exception: {exception}")
        logger.error(traceback.format_exc())
        
        # В development показываем детали
        if settings.DEBUG:
            return None  # Показываем стандартный Django error page
        
        # В production отправляем JSON ошибку
        return JsonResponse({
            'error': 'Internal server error',
            'message': 'Something went wrong. Please try again later.'
        }, status=500)
```

## 2. FastAPI Middleware

### Middleware в FastAPI

```python
from fastapi import FastAPI, Request
import time
import logging

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

# Пример 1: Логирование времени выполнения запроса
@app.middleware("http")
async def log_request_time(request: Request, call_next):
    start_time = time.time()
    
    # Выполняем запрос
    response = await call_next(request)
    
    # Вычисляем время
    process_time = time.time() - start_time
    logger.info(f"{request.method} {request.url.path} - {process_time:.3f}s")
    
    # Добавляем заголовок с временем выполнения
    response.headers["X-Process-Time"] = str(process_time)
    
    return response

# Пример 2: Добавление correlation ID для отслеживания
@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
    import uuid
    
    # Генерируем уникальный ID для этого запроса
    correlation_id = str(uuid.uuid4())
    
    # Добавляем в контекст запроса
    request.state.correlation_id = correlation_id
    
    # Выполняем запрос
    response = await call_next(request)
    
    # Добавляем ID в ответ
    response.headers["X-Correlation-ID"] = correlation_id
    
    return response
```

### Пример 3: CORS Middleware

```python
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

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

# Или с переменной окружения
import os

allowed_origins = os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(",")

app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
```

## 3. ASGI Middleware

### Пример: Custom ASGI Middleware для валидации токена

```python
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import jwt

app = FastAPI()

class JWTMiddleware:
    """Middleware для проверки JWT токена"""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        
        # Получаем заголовок Authorization
        headers = dict(scope["headers"])
        auth_header = headers.get(b"authorization", b"").decode()
        
        # Проверяем формат
        if not auth_header.startswith("Bearer "):
            response = JSONResponse(
                {"detail": "Invalid authorization header"},
                status_code=401
            )
            await response(scope, receive, send)
            return
        
        # Извлекаем токен
        token = auth_header.split(" ")[1]
        
        # Проверяем токен
        try:
            payload = jwt.decode(token, "secret", algorithms=["HS256"])
            scope["user"] = payload
        except jwt.InvalidTokenError:
            response = JSONResponse(
                {"detail": "Invalid token"},
                status_code=401
            )
            await response(scope, receive, send)
            return
        
        # Передаём дальше
        await self.app(scope, receive, send)

app.add_middleware(JWTMiddleware)
```

## 4. Регистрация Middleware в Django

```python
# settings.py

MIDDLEWARE = [
    # Django встроенный middleware
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    
    # Ваш кастомный middleware
    'myapp.middleware.RequestLoggingMiddleware',
    'myapp.middleware.SecurityHeadersMiddleware',
    'myapp.middleware.CachingMiddleware',
    'myapp.middleware.ErrorHandlingMiddleware',
]
```

## 5. Реальный практический пример: Полный middleware

```python
# middleware.py
import logging
import time
from django.utils.deprecation import MiddlewareMixin
from django.http import JsonResponse
import json
from django.conf import settings

logger = logging.getLogger(__name__)

class ComprehensiveLoggingMiddleware(MiddlewareMixin):
    """
    Полный middleware для логирования, мониторинга и обработки ошибок
    """
    
    def process_request(self, request):
        # Сохраняем время начала
        request._start_time = time.time()
        
        # Логируем запрос
        logger.info(f"
{'='*80}")
        logger.info(f"REQUEST: {request.method} {request.path}")
        logger.info(f"IP: {self.get_client_ip(request)}")
        logger.info(f"User: {request.user}")
        
        if request.method in ['POST', 'PUT', 'PATCH']:
            try:
                logger.info(f"Body: {request.body.decode()[:500]}")
            except:
                pass
    
    def process_response(self, request, response):
        # Вычисляем время выполнения
        if hasattr(request, '_start_time'):
            duration = time.time() - request._start_time
            logger.info(f"RESPONSE: {response.status_code} - {duration:.3f}s")
        
        logger.info(f"{'='*80}
")
        
        return response
    
    def process_exception(self, request, exception):
        logger.error(f"EXCEPTION: {exception}")
        
        if settings.DEBUG:
            return None
        
        return JsonResponse(
            {'error': 'Internal server error'},
            status=500
        )
    
    @staticmethod
    def get_client_ip(request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0]
        return request.META.get('REMOTE_ADDR')
```

## Резюме

Middleware используется для:
- **Логирования** — отслеживание запросов и ошибок
- **Аутентификации** — проверка прав доступа
- **Кэширования** — улучшение производительности
- **Безопасности** — добавление заголовков и проверок
- **Обработки ошибок** — красивая обработка исключений
- **Мониторинга** — измерение производительности
- **Трансформации данных** — изменение request/response

Выбор middleware зависит от вашего фреймворка (Django, FastAPI, Flask), но принцип везде одинаков: перехватить запрос, обработать, и передать дальше.

Приведи пример использования middleware

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

Middleware в Python: Примеры и Использование

1. Django Middleware

Что такое Django Middleware?

Пример 1: Логирование всех запросов

Пример 2: Добавление заголовков безопасности

Пример 3: Аутентификация и авторизация

Пример 4: Кэширование ответов

Пример 5: Обработка ошибок и исключений

2. FastAPI Middleware

Middleware в FastAPI

Пример 3: CORS Middleware

3. ASGI Middleware

Пример: Custom ASGI Middleware для валидации токена

4. Регистрация Middleware в Django

5. Реальный практический пример: Полный middleware

Резюме