Просунуте Middleware в Django 2026: Створення Власних Middleware, Логування та Питання на Співбесідах

Повний посібник з просунутих технік middleware в Django 2026. Дізнайтеся, як створювати власні middleware, імплементувати логування та підготуватися до технічних співбесід.

Просунуте middleware в Django 2026

Django middleware є одним із найважливіших компонентів архітектури фреймворку, виконуючи роль проміжного шару між HTTP-запитом та представленням (view) додатку. У 2026 році middleware залишається ключовим інструментом для розробників Django, що дозволяє реалізовувати наскрізну функціональність без модифікації бізнес-логіки представлень.

Middleware в Django працює за принципом цибулини — кожен шар обробляє запит на шляху до представлення, а потім обробляє відповідь на зворотному шляху до клієнта. Розуміння цього потоку є критичним для ефективного використання middleware.

Архітектура Middleware в Django

Middleware в Django — це класи або функції, які викликаються для кожного HTTP-запиту, що проходить через додаток. Система middleware працює за патерном ланцюга відповідальності, де кожне middleware може:

  • Обробити запит перед передачею наступному middleware або представленню
  • Обробити відповідь перед поверненням клієнту
  • Перервати ланцюг та повернути власну відповідь
  • Обробити винятки, згенеровані представленнями або іншими middleware
python
# settings.py - Порядок middleware має значення
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',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    # Власні middleware зазвичай додаються в кінці
    'myapp.middleware.CustomMiddleware',
]

Створення Власного Middleware в Класовому Стилі

Django підтримує два стилі визначення middleware: класовий та функціональний. Класовий стиль пропонує більшу гнучкість і є переважним для складних випадків використання.

python
# myapp/middleware.py
from django.http import HttpRequest, HttpResponse
from django.utils.deprecation import MiddlewareMixin
import time
import logging

logger = logging.getLogger(__name__)


class RequestTimingMiddleware:
    """Middleware для вимірювання часу виконання запиту."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        # Одноразова конфігурація, що виконується при старті сервера
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # Код, що виконується перед представленням
        start_time = time.perf_counter()
        
        # Виклик наступного middleware або представлення
        response = self.get_response(request)
        
        # Код, що виконується після представлення
        duration = time.perf_counter() - start_time
        response['X-Request-Duration'] = f'{duration:.4f}s'
        
        logger.info(
            f'Request {request.method} {request.path} '
            f'completed in {duration:.4f}s'
        )
        
        return response

Middleware з Хуками для Обробки Представлень та Винятків

Для складних випадків Django надає спеціальні методи-хуки, які дозволяють точно контролювати різні етапи обробки запиту.

python
# myapp/middleware.py
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.core.exceptions import PermissionDenied
import traceback
import logging

logger = logging.getLogger(__name__)


class AdvancedMiddleware:
    """Просунуте middleware з повним набором хуків."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # Попередня обробка запиту
        self.process_request(request)
        
        response = self.get_response(request)
        
        # Постобробка відповіді
        return self.process_response(request, response)
    
    def process_request(self, request: HttpRequest) -> None:
        """Обробка перед передачею до представлення."""
        request.custom_data = {
            'middleware_processed': True,
            'timestamp': time.time()
        }
    
    def process_response(
        self, 
        request: HttpRequest, 
        response: HttpResponse
    ) -> HttpResponse:
        """Обробка після виконання представлення."""
        response['X-Processed-By'] = 'AdvancedMiddleware'
        return response
    
    def process_view(
        self, 
        request: HttpRequest, 
        view_func, 
        view_args, 
        view_kwargs
    ):
        """Викликається безпосередньо перед викликом представлення."""
        logger.debug(f'Calling view: {view_func.__name__}')
        return None  # None означає продовження обробки
    
    def process_exception(
        self, 
        request: HttpRequest, 
        exception: Exception
    ):
        """Обробка винятків, згенерованих представленням."""
        if isinstance(exception, PermissionDenied):
            return JsonResponse(
                {'error': 'Access denied'},
                status=403
            )
        
        logger.error(
            f'Unhandled exception: {exception}\n'
            f'{traceback.format_exc()}'
        )
        return None  # Передача до стандартної обробки
    
    def process_template_response(
        self, 
        request: HttpRequest, 
        response
    ):
        """Обробка шаблонних відповідей."""
        if hasattr(response, 'context_data'):
            response.context_data['middleware_info'] = 'Processed'
        return response

Імплементація Просунутої Системи Логування

Логування є одним із найпоширеніших застосувань middleware. Професійна імплементація повинна враховувати структуроване логування, кореляцію запитів та продуктивність.

python
# myapp/middleware/logging.py
import uuid
import json
import time
import logging
from django.http import HttpRequest, HttpResponse
from django.conf import settings

logger = logging.getLogger('request_logger')


class StructuredLoggingMiddleware:
    """Middleware для структурованого логування HTTP-запитів."""
    
    SENSITIVE_HEADERS = {'authorization', 'cookie', 'x-api-key'}
    EXCLUDED_PATHS = {'/health/', '/metrics/', '/favicon.ico'}
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # Генерування унікального ID запиту
        request_id = str(uuid.uuid4())
        request.request_id = request_id
        
        # Пропуск логування для виключених шляхів
        if request.path in self.EXCLUDED_PATHS:
            return self.get_response(request)
        
        start_time = time.perf_counter()
        
        # Логування запиту
        self._log_request(request, request_id)
        
        try:
            response = self.get_response(request)
            duration = time.perf_counter() - start_time
            
            # Логування відповіді
            self._log_response(request, response, request_id, duration)
            
            # Додавання ID запиту до заголовків відповіді
            response['X-Request-ID'] = request_id
            
            return response
            
        except Exception as e:
            duration = time.perf_counter() - start_time
            self._log_error(request, e, request_id, duration)
            raise
    
    def _log_request(self, request: HttpRequest, request_id: str) -> None:
        """Логування вхідного запиту."""
        log_data = {
            'event': 'request_started',
            'request_id': request_id,
            'method': request.method,
            'path': request.path,
            'query_params': dict(request.GET),
            'user_agent': request.META.get('HTTP_USER_AGENT', ''),
            'ip_address': self._get_client_ip(request),
            'user_id': getattr(request.user, 'id', None),
            'headers': self._get_safe_headers(request),
        }
        
        logger.info(json.dumps(log_data))
    
    def _log_response(
        self, 
        request: HttpRequest, 
        response: HttpResponse,
        request_id: str,
        duration: float
    ) -> None:
        """Логування відповіді."""
        log_data = {
            'event': 'request_completed',
            'request_id': request_id,
            'method': request.method,
            'path': request.path,
            'status_code': response.status_code,
            'duration_ms': round(duration * 1000, 2),
            'content_length': len(response.content) if hasattr(response, 'content') else 0,
        }
        
        log_level = logging.WARNING if response.status_code >= 400 else logging.INFO
        logger.log(log_level, json.dumps(log_data))
    
    def _log_error(
        self, 
        request: HttpRequest, 
        exception: Exception,
        request_id: str,
        duration: float
    ) -> None:
        """Логування помилки."""
        log_data = {
            'event': 'request_failed',
            'request_id': request_id,
            'method': request.method,
            'path': request.path,
            'error_type': type(exception).__name__,
            'error_message': str(exception),
            'duration_ms': round(duration * 1000, 2),
        }
        
        logger.error(json.dumps(log_data), exc_info=True)
    
    def _get_client_ip(self, request: HttpRequest) -> str:
        """Отримання IP-адреси клієнта з урахуванням проксі."""
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0].strip()
        return request.META.get('REMOTE_ADDR', '')
    
    def _get_safe_headers(self, request: HttpRequest) -> dict:
        """Отримання заголовків з приховуванням чутливих даних."""
        headers = {}
        for key, value in request.META.items():
            if key.startswith('HTTP_'):
                header_name = key[5:].lower().replace('_', '-')
                if header_name in self.SENSITIVE_HEADERS:
                    headers[header_name] = '[REDACTED]'
                else:
                    headers[header_name] = value
        return headers

Middleware для Контролю Доступу та Rate Limiting

Контроль доступу та обмеження кількості запитів є критичними функціями для продакшн-додатків.

python
# myapp/middleware/rate_limiting.py
from django.http import HttpRequest, JsonResponse
from django.core.cache import cache
from django.conf import settings
import time


class RateLimitMiddleware:
    """Middleware для обмеження кількості запитів на користувача/IP."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        self.rate_limit = getattr(settings, 'RATE_LIMIT_REQUESTS', 100)
        self.rate_window = getattr(settings, 'RATE_LIMIT_WINDOW', 60)
        
    def __call__(self, request: HttpRequest):
        # Ідентифікація клієнта
        client_id = self._get_client_identifier(request)
        cache_key = f'rate_limit:{client_id}'
        
        # Отримання поточної кількості запитів
        request_data = cache.get(cache_key, {'count': 0, 'reset_at': 0})
        current_time = time.time()
        
        # Скидання лічильника, якщо часове вікно закінчилось
        if current_time > request_data['reset_at']:
            request_data = {
                'count': 0,
                'reset_at': current_time + self.rate_window
            }
        
        # Перевірка ліміту
        if request_data['count'] >= self.rate_limit:
            retry_after = int(request_data['reset_at'] - current_time)
            return JsonResponse(
                {
                    'error': 'Rate limit exceeded',
                    'retry_after': retry_after
                },
                status=429,
                headers={'Retry-After': str(retry_after)}
            )
        
        # Інкрементація лічильника
        request_data['count'] += 1
        cache.set(cache_key, request_data, self.rate_window)
        
        response = self.get_response(request)
        
        # Додавання інформаційних заголовків
        response['X-RateLimit-Limit'] = str(self.rate_limit)
        response['X-RateLimit-Remaining'] = str(
            self.rate_limit - request_data['count']
        )
        response['X-RateLimit-Reset'] = str(int(request_data['reset_at']))
        
        return response
    
    def _get_client_identifier(self, request: HttpRequest) -> str:
        """Ідентифікація клієнта на основі користувача або IP."""
        if request.user.is_authenticated:
            return f'user:{request.user.id}'
        
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            ip = x_forwarded_for.split(',')[0].strip()
        else:
            ip = request.META.get('REMOTE_ADDR', 'unknown')
        
        return f'ip:{ip}'

Асинхронне Middleware в Django

Django версії 4.1 та новіших підтримує асинхронне middleware, що особливо корисно для додатків, які використовують ASGI.

python
# myapp/middleware/async_middleware.py
from django.http import HttpRequest, HttpResponse
import asyncio
import aiohttp
import logging

logger = logging.getLogger(__name__)


class AsyncExternalServiceMiddleware:
    """Асинхронне middleware для інтеграції із зовнішнім сервісом."""
    
    async_capable = True
    sync_capable = False
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    async def __call__(self, request: HttpRequest) -> HttpResponse:
        # Асинхронне отримання даних із зовнішнього сервісу
        try:
            external_data = await self._fetch_external_data(request)
            request.external_data = external_data
        except Exception as e:
            logger.warning(f'Failed to fetch external data: {e}')
            request.external_data = None
        
        response = await self.get_response(request)
        
        return response
    
    async def _fetch_external_data(self, request: HttpRequest) -> dict:
        """Асинхронне отримання даних із зовнішнього API."""
        async with aiohttp.ClientSession() as session:
            async with session.get(
                'https://api.example.com/data',
                timeout=aiohttp.ClientTimeout(total=2)
            ) as response:
                if response.status == 200:
                    return await response.json()
                return {}


class HybridMiddleware:
    """Middleware, що підтримує як sync, так і async."""
    
    async_capable = True
    sync_capable = True
    
    def __init__(self, get_response):
        self.get_response = get_response
        # Визначення, чи get_response є async
        if asyncio.iscoroutinefunction(get_response):
            self._is_async = True
        else:
            self._is_async = False
    
    def __call__(self, request: HttpRequest) -> HttpResponse:
        if self._is_async:
            return self._async_call(request)
        return self._sync_call(request)
    
    def _sync_call(self, request: HttpRequest) -> HttpResponse:
        request.middleware_mode = 'sync'
        return self.get_response(request)
    
    async def _async_call(self, request: HttpRequest) -> HttpResponse:
        request.middleware_mode = 'async'
        return await self.get_response(request)

Тестування Middleware

Тестування middleware вимагає особливого підходу через його позицію в ланцюзі обробки запитів.

python
# tests/test_middleware.py
from django.test import TestCase, RequestFactory, override_settings
from django.contrib.auth import get_user_model
from django.http import HttpResponse
from myapp.middleware import RequestTimingMiddleware, RateLimitMiddleware
from unittest.mock import patch, MagicMock

User = get_user_model()


class RequestTimingMiddlewareTest(TestCase):
    """Тести для RequestTimingMiddleware."""
    
    def setUp(self):
        self.factory = RequestFactory()
        self.get_response = MagicMock(return_value=HttpResponse('OK'))
        self.middleware = RequestTimingMiddleware(self.get_response)
    
    def test_adds_duration_header(self):
        """Тест, чи middleware додає заголовок з часом виконання."""
        request = self.factory.get('/test/')
        
        response = self.middleware(request)
        
        self.assertIn('X-Request-Duration', response)
        self.assertTrue(response['X-Request-Duration'].endswith('s'))
    
    def test_calls_get_response(self):
        """Тест, чи middleware викликає наступний елемент ланцюга."""
        request = self.factory.get('/test/')
        
        self.middleware(request)
        
        self.get_response.assert_called_once_with(request)
    
    @patch('myapp.middleware.logger')
    def test_logs_request_info(self, mock_logger):
        """Тест, чи middleware логує інформацію про запит."""
        request = self.factory.get('/test/')
        
        self.middleware(request)
        
        mock_logger.info.assert_called_once()
        log_message = mock_logger.info.call_args[0][0]
        self.assertIn('GET', log_message)
        self.assertIn('/test/', log_message)


class RateLimitMiddlewareTest(TestCase):
    """Тести для RateLimitMiddleware."""
    
    def setUp(self):
        self.factory = RequestFactory()
        self.get_response = MagicMock(return_value=HttpResponse('OK'))
    
    @override_settings(RATE_LIMIT_REQUESTS=5, RATE_LIMIT_WINDOW=60)
    def test_allows_requests_within_limit(self):
        """Тест, чи middleware пропускає запити в межах ліміту."""
        middleware = RateLimitMiddleware(self.get_response)
        request = self.factory.get('/test/')
        request.user = MagicMock(is_authenticated=False)
        request.META['REMOTE_ADDR'] = '192.168.1.1'
        
        for _ in range(5):
            response = middleware(request)
            self.assertEqual(response.status_code, 200)
    
    @override_settings(RATE_LIMIT_REQUESTS=2, RATE_LIMIT_WINDOW=60)
    def test_blocks_requests_over_limit(self):
        """Тест, чи middleware блокує запити, що перевищують ліміт."""
        middleware = RateLimitMiddleware(self.get_response)
        request = self.factory.get('/test/')
        request.user = MagicMock(is_authenticated=False)
        request.META['REMOTE_ADDR'] = '192.168.1.2'
        
        # Перші два запити повинні пройти
        middleware(request)
        middleware(request)
        
        # Третій запит має бути заблокований
        response = middleware(request)
        self.assertEqual(response.status_code, 429)

Готовий до співбесід з Django?

Практикуйся з нашими інтерактивними симуляторами, flashcards та технічними тестами.

Відкрий Django

Питання на Співбесідах щодо Middleware в Django

Під час технічних співбесід на позиції, пов'язані з Django, часто з'являються питання про middleware. Ось найважливіші теми з відповідями.

Питання: Яка різниця між MIDDLEWARE та MIDDLEWARE_CLASSES в Django?

MIDDLEWARE_CLASSES — це стара конфігурація, що використовувалась у Django < 1.10 та вимагала успадкування від MiddlewareMixin. MIDDLEWARE — це новий стиль, впроваджений у Django 1.10, що використовує простіший патерн з методом __call__ та параметром get_response. Новий стиль більш гнучкий і підтримує як синхронне, так і асинхронне middleware.

Питання: В якому порядку виконуються middleware під час обробки запиту та відповіді?

Middleware виконуються в порядку, визначеному в MIDDLEWARE, під час обробки запиту (зверху вниз), а потім у зворотному порядку під час обробки відповіді (знизу вгору). Ця архітектура нагадує цибулину або стек LIFO.

Питання: Як перервати ланцюг middleware та повернути відповідь раніше?

Щоб перервати ланцюг, middleware повинне повернути об'єкт HttpResponse безпосередньо в методі __call__ замість виклику get_response. Прикладом є middleware для перевірки авторизації, яке повертає відповідь 403 без подальшої обробки.

Питання: Коли використовується process_view, а коли process_exception?

process_view викликається безпосередньо перед викликом представлення і має доступ до функції представлення та її аргументів. Може використовуватись для контролю доступу або модифікації аргументів. process_exception викликається тільки коли представлення генерує виняток і служить для централізованої обробки помилок.

Питання: Як імплементувати middleware, що підтримує як WSGI, так і ASGI?

Потрібно встановити атрибути класу sync_capable = True та async_capable = True, а потім у конструкторі визначити, чи get_response є асинхронною функцією. У методі __call__ потрібно відповідно викликати sync або async версію обробки.

Найкращі Практики та Патерни

При імплементації middleware варто дотримуватись кількох принципів:

  1. Принцип єдиної відповідальності — кожне middleware повинне виконувати одну, чітко визначену функцію
  2. Продуктивність — middleware викликається для кожного запиту, тому операції мають бути максимально оптимізовані
  3. Стійкість до помилок — middleware не повинне спричиняти збій додатку; винятки потрібно належним чином обробляти
  4. Конфігурованість — параметри middleware мають бути налаштовуваними через settings.py
  5. Тестованість — middleware повинне легко тестуватись ізольовано

Підсумок

Middleware в Django — це потужний інструмент, що дозволяє реалізовувати наскрізну функціональність модульним та повторно використовуваним способом. Розуміння архітектури middleware, вміння створювати власні імплементації та знання найкращих практик є необхідними для кожного просунутого розробника Django. Представлені в статті приклади охоплюють найпоширеніші випадки використання: логування, контроль доступу, rate limiting та асинхронну обробку, створюючи міцну основу для подальшого розвитку власних рішень.

Щоденний виклик

Чи знайдеш ти помилку в Django?

Справжній фрагмент коду, прихована помилка, одна спроба на день. Щоб спробувати, акаунт не потрібен.

Прийняти виклик
Anthony Fillion-Maillet

Автор:

Anthony Fillion-Maillet

Засновник SharpSkill

Fullstack-розробник понад 10 років. Керує SharpSkill і відповідає за все, що тут публікується.

Оновлено 18 вересня 2026 р.

Теги

#django
#middleware
#python
#логування
#backend

Поділитися

Пов'язані статті

Django Signals vs Celery Tasks 2026

Django Signals vs Celery Tasks у 2026: Коли Використовувати та Питання на Співбесіді

Комплексне порівняння Django Signals та Celery Tasks у 2026 році. Коли використовувати сигнали, а коли асинхронні задачі, з прикладами коду та питаннями на технічну співбесіду.

Django 5.2 middleware та сигнали: архітектурна діаграма обробки HTTP-запитів і подій

Django 5.2: Власна Middleware та Обробка Сигналів для Технічних Співбесід

Повний посібник з Django 5.2: побудова власної middleware, використання сигналів post_save та pre_save, асинхронна middleware, користувацькі сигнали та типові питання для технічних співбесід з прикладами коду.

Підготовка до співбесіди з Django: ORM, Middleware та Django REST Framework

Питання на співбесіді з Django: ORM, Middleware та DRF, поглиблений розбір

Питання на співбесіді з Django: оптимізація ORM з select_related, prefetch_related та режимом FETCH_PEERS у Django 6.1, архітектура middleware, продуктивність серіалізаторів Django REST Framework, дозволи та пагінація.