Создание middleware, порядок выполнения, request/response hooks
Middleware — это слои-обёртки вокруг view, которые обрабатывают запросы и ответы глобально для всего приложения.
💡 Правило: Используйте middleware для кросс-режущей логики: логирование, аутентификация, security headers. Не используйте для бизнес-логики конкретного приложения.
Request → M1 → M2 → M3 → View → M3 → M2 → M1 → Response
↓ ↓ ↓ ↓ ↓ ↓ ↓
process_request... process_response...
# settings.py
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware', # 1. Запрос вниз, 6. Ответ вверх
'django.contrib.sessions.middleware.SessionMiddleware', # 2. Запрос вниз, 5. Ответ вверх
'django.middleware.common.CommonMiddleware', # 3. Запрос вниз, 4. Ответ вверх
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
]# middleware.py
import time
from django.utils.deprecation import MiddlewareMixin
class SimpleMiddleware:
"""Базовый middleware без наследования."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Код ДО view (request фаза)
print(f"Request: {request.method} {request.path}")
request.start_time = time.time()
# Вызов следующего middleware или view
response = self.get_response(request)
# Код ПОСЛЕ view (response фаза)
duration = time.time() - request.start_time
print(f"Response: {response.status_code} in {duration:.3f}s")
return responsefrom django.utils.deprecation import MiddlewareMixin
class LegacyMiddleware(MiddlewareMixin):
"""Middleware со старыми хуками."""
def process_request(self, request):
"""Вызывается ДО view. Вернуть None или HttpResponse."""
print(f"process_request: {request.path}")
return None # Продолжить обработку
def process_view(self, request, view_func, view_args, view_kwargs):
"""Вызывается ПЕРЕД вызовом view. Вернуть None или HttpResponse."""
print(f"process_view: {view_func.__name__}")
return None
def process_template_response(self, request, response):
"""Вызывается ПОСЛЕ view для TemplateResponse."""
print(f"process_template_response: {response.template_name}")
return response
def process_response(self, request, response):
"""Вызывается ПОСЛЕ view. Вернуть HttpResponse."""
print(f"process_response: {response.status_code}")
return response
def process_exception(self, request, exception):
"""Вызывается если view выбросил исключение."""
print(f"process_exception: {exception}")
return None # Продолжить обработку ошибки# settings.py
MIDDLEWARE = [
# ...
'myapp.middleware.SimpleMiddleware',
'myapp.middleware.LegacyMiddleware',
]# settings.py
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware', # Должен быть первым!
# ...
]
# Настройки безопасности
SECURE_SSL_REDIRECT = True # HTTPS редирект
SECURE_HSTS_SECONDS = 31536000 # HSTS (1 год)
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SECURE_CONTENT_TYPE_NOSNIFF = True # X-Content-Type-Options
SECURE_BROWSER_XSS_FILTER = True # X-XSS-Protection (устарело)
X_FRAME_OPTIONS = 'DENY' # Clickjacking protection# Добавляет request.session
MIDDLEWARE = [
'django.contrib.sessions.middleware.SessionMiddleware',
# ...
]
# Использование
request.session['key'] = 'value'
value = request.session.get('key')
del request.session['key']# Добавляет request.user
MIDDLEWARE = [
'django.contrib.sessions.middleware.SessionMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware', # После SessionMiddleware
# ...
]
# Использование
if request.user.is_authenticated:
print(f"Hello, {request.user.username}")
else:
print("Guest")# Защищает от CSRF атак
MIDDLEWARE = [
'django.middleware.csrf.CsrfViewMiddleware',
# ...
]
# В шаблонах обязательно
<form method="post">
{% csrf_token %}
<!-- поля формы -->
</form>
# Для API endpoints (если нужно отключить)
from django.views.decorators.csrf import csrf_exempt
@csrf_exempt
def api_endpoint(request):
return JsonResponse({'status': 'ok'})# Разное: APPEND_SLASH, ETag, gzip, etc.
MIDDLEWARE = [
'django.middleware.common.CommonMiddleware',
# ...
]
# Настройки
APPEND_SLASH = True # Добавляет / к URL без него
USE_ETAGS = True # Добавляет ETag header# Добавляет сообщения для пользователя
MIDDLEWARE = [
'django.contrib.messages.middleware.MessageMiddleware',
# ...
]
# В views
from django.contrib import messages
def my_view(request):
messages.success(request, 'Operation completed!')
messages.error(request, 'Something went wrong.')
return redirect('success')
# В template
{% for message in messages %}
<div class="alert alert-{{ message.tags }}">{{ message }}</div>
{% endfor %}# middleware/logging.py
import logging
import time
import uuid
logger = logging.getLogger(__name__)
class RequestLoggingMiddleware:
"""Логгирует все запросы с временем выполнения."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Генерируем ID запроса
request.request_id = str(uuid.uuid4())
# Логируем запрос
logger.info(
f"Request [{request.request_id}]: "
f"{request.method} {request.path} "
f"from {request.META.get('REMOTE_ADDR')}"
)
# Засекаем время
start_time = time.time()
# Получаем ответ
response = self.get_response(request)
# Логируем ответ
duration = time.time() - start_time
logger.info(
f"Response [{request.request_id}]: "
f"{response.status_code} in {duration:.3f}s"
)
# Добавляем ID запроса в заголовок
response['X-Request-ID'] = request.request_id
return response# middleware/bot_blocker.py
from django.http import HttpResponseForbidden
BLOCKED_BOTS = [
'badbot',
'spambot',
'scraper',
]
class BotBlockerMiddleware:
"""Блокирует нежелательных ботов."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
user_agent = request.META.get('HTTP_USER_AGENT', '').lower()
for bot in BLOCKED_BOTS:
if bot.lower() in user_agent:
logger.warning(f"Blocked bot: {user_agent}")
return HttpResponseForbidden('Bots not allowed')
return self.get_response(request)# middleware/domain_locale.py
from django.utils import translation
class DomainLocaleMiddleware:
"""Устанавливает язык по домену."""
def __init__(self, get_response):
self.get_response = get_response
DOMAIN_LANGUAGE_MAP = {
'example.com': 'en',
'example.ru': 'ru',
'example.de': 'de',
}
def __call__(self, request):
host = request.get_host().split(':')[0] # Без порта
language = self.DOMAIN_LANGUAGE_MAP.get(host)
if language:
translation.activate(language)
request.LANGUAGE_CODE = language
response = self.get_response(request)
if language:
translation.deactivate()
return response# middleware/ratelimit.py
import time
from django.core.cache import cache
from django.http import HttpResponseTooManyRequests
class RateLimitMiddleware:
"""Ограничивает количество запросов."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Получаем IP пользователя
ip = self.get_client_ip(request)
# Ключ для кэша
key = f'ratelimit:{ip}'
# Получаем текущее количество запросов
requests = cache.get(key, 0)
# Лимит: 100 запросов в минуту
if requests >= 100:
return HttpResponseTooManyRequests('Too many requests')
# Увеличиваем счётчик
cache.set(key, requests + 1, 60) # TTL 60 секунд
response = self.get_response(request)
# Добавляем заголовки rate limit
response['X-RateLimit-Limit'] = '100'
response['X-RateLimit-Remaining'] = str(100 - requests - 1)
return response
def get_client_ip(self, request):
xff = request.META.get('HTTP_X_FORWARDED_FOR')
if xff:
return xff.split(',')[0].strip()
return request.META.get('REMOTE_ADDR')# middleware/maintenance.py
from django.conf import settings
from django.http import HttpResponseServiceUnavailable
from django.template import loader
class MaintenanceModeMiddleware:
"""Режим обслуживания."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Проверяем режим обслуживания
if getattr(settings, 'MAINTENANCE_MODE', False):
# Разрешаем доступ админам
if not (request.user.is_authenticated and request.user.is_staff):
# Разрешаем доступ к admin и media
allowed_paths = ['/admin/', '/media/', '/static/']
if not any(request.path.startswith(p) for p in allowed_paths):
template = loader.get_template('maintenance.html')
return HttpResponseServiceUnavailable(template.render({}))
return self.get_response(request)# middleware/error_handler.py
import json
from django.http import JsonResponse
from django.template import loader
from django.http import Http404, PermissionDenied
class ErrorHandlerMiddleware:
"""Кастомная обработка ошибок."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
return self.get_response(request)
def process_exception(self, request, exception):
"""Обрабатывает исключения из view."""
# 404 Not Found
if isinstance(exception, Http404):
if request.path.startswith('/api/'):
return JsonResponse(
{'error': 'Not Found', 'status': 404},
status=404
)
template = loader.get_template('404.html')
return HttpResponseNotFound(template.render({'path': request.path}))
# 403 Permission Denied
if isinstance(exception, PermissionDenied):
if request.path.startswith('/api/'):
return JsonResponse(
{'error': 'Forbidden', 'status': 403},
status=403
)
template = loader.get_template('403.html')
return HttpResponseForbidden(template.render({}))
# 500 Server Error
# Логируем ошибку
import logging
logger = logging.getLogger(__name__)
logger.error(f'Server error: {exception}', exc_info=True)
if request.path.startswith('/api/'):
return JsonResponse(
{'error': 'Internal Server Error', 'status': 500},
status=500
)
# Для обычных запросов возвращаем None (использовать стандартную 500)
return None# middleware/__init__.py
from .request_logging import RequestLoggingMiddleware
from .security_headers import SecurityHeadersMiddleware
from .rate_limit import RateLimitMiddleware
from .maintenance import MaintenanceModeMiddleware
from .error_handler import ErrorHandlerMiddleware
__all__ = [
'RequestLoggingMiddleware',
'SecurityHeadersMiddleware',
'RateLimitMiddleware',
'MaintenanceModeMiddleware',
'ErrorHandlerMiddleware',
]# middleware/security_headers.py
class SecurityHeadersMiddleware:
"""Добавляет security headers."""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
# Content Security Policy
response['Content-Security-Policy'] = (
"default-src 'self'; "
"script-src 'self' 'unsafe-inline'; "
"style-src 'self' 'unsafe-inline'; "
"img-src 'self' data: https:; "
"font-src 'self'; "
"frame-ancestors 'none';"
)
# Referrer Policy
response['Referrer-Policy'] = 'strict-origin-when-cross-origin'
# Permissions Policy
response['Permissions-Policy'] = (
'geolocation=(), microphone=(), camera=()'
)
# Cross-Origin Policies
response['Cross-Origin-Opener-Policy'] = 'same-origin'
response['Cross-Origin-Resource-Policy'] = 'same-origin'
response['Cross-Origin-Embedder-Policy'] = 'require-corp'
return response# settings.py
MIDDLEWARE = [
# Security (должен быть первым)
'django.middleware.security.SecurityMiddleware',
'myapp.middleware.security_headers.SecurityHeadersMiddleware',
# Sessions & Auth
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
# Custom middleware
'myapp.middleware.request_logging.RequestLoggingMiddleware',
'myapp.middleware.rate_limit.RateLimitMiddleware',
'myapp.middleware.maintenance.MaintenanceModeMiddleware',
'myapp.middleware.error_handler.ErrorHandlerMiddleware',
# Clickjacking (должен быть последним для response)
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
# Настройки
MAINTENANCE_MODE = False # Включить режим обслуживания
RATE_LIMIT_REQUESTS = 100 # Запросов в минуту
RATE_LIMIT_WINDOW = 60 # Окно в секундахВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.