Zaawansowane Middleware w Django 2026: Tworzenie Własnych Middleware, Logowanie i Pytania Rekrutacyjne

Kompleksowy przewodnik po zaawansowanych technikach middleware w Django 2026. Dowiedz się, jak tworzyć własne middleware, implementować logowanie i przygotować się do rozmów kwalifikacyjnych.

Zaawansowane middleware w Django 2026

Django middleware stanowi jeden z najważniejszych elementów architektury frameworka, działając jako warstwa pośrednia pomiędzy żądaniem HTTP a widokiem aplikacji. W 2026 roku middleware pozostaje kluczowym narzędziem dla programistów Django, umożliwiającym implementację przekrojowych funkcjonalności bez modyfikacji logiki biznesowej widoków.

Middleware w Django działa w sposób podobny do cebuli — każda warstwa przetwarza żądanie w drodze do widoku, a następnie przetwarza odpowiedź w drodze powrotnej do klienta. Zrozumienie tego przepływu jest kluczowe dla efektywnego wykorzystania middleware.

Architektura Middleware w Django

Middleware w Django to klasy lub funkcje, które są wywoływane dla każdego żądania HTTP przechodzącego przez aplikację. System middleware działa według wzorca łańcucha odpowiedzialności, gdzie każdy middleware może:

  • Przetworzyć żądanie przed przekazaniem go do następnego middleware lub widoku
  • Przetworzyć odpowiedź przed zwróceniem jej do klienta
  • Przerwać łańcuch i zwrócić własną odpowiedź
  • Obsłużyć wyjątki zgłoszone przez widoki lub inne middleware
python
# settings.py - Kolejność middleware ma znaczenie
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',
    # Własne middleware dodaje się zazwyczaj na końcu
    'myapp.middleware.CustomMiddleware',
]

Tworzenie Własnego Middleware w Stylu Klasowym

Django wspiera dwa style definiowania middleware: klasowy i funkcyjny. Styl klasowy oferuje większą elastyczność i jest preferowany dla złożonych przypadków użycia.

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 mierzący czas wykonania żądania."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        # Jednorazowa konfiguracja wykonywana przy starcie serwera
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # Kod wykonywany przed widokiem
        start_time = time.perf_counter()
        
        # Wywołanie następnego middleware lub widoku
        response = self.get_response(request)
        
        # Kod wykonywany po widoku
        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 z Hookami do Obsługi Widoków i Wyjątków

Dla zaawansowanych przypadków Django udostępnia specjalne metody hook, które pozwalają na precyzyjną kontrolę nad różnymi etapami przetwarzania żądania.

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:
    """Zaawansowane middleware z pełnym zestawem hooków."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # Pre-processing żądania
        self.process_request(request)
        
        response = self.get_response(request)
        
        # Post-processing odpowiedzi
        return self.process_response(request, response)
    
    def process_request(self, request: HttpRequest) -> None:
        """Przetwarzanie przed przekazaniem do widoku."""
        request.custom_data = {
            'middleware_processed': True,
            'timestamp': time.time()
        }
    
    def process_response(
        self, 
        request: HttpRequest, 
        response: HttpResponse
    ) -> HttpResponse:
        """Przetwarzanie po wykonaniu widoku."""
        response['X-Processed-By'] = 'AdvancedMiddleware'
        return response
    
    def process_view(
        self, 
        request: HttpRequest, 
        view_func, 
        view_args, 
        view_kwargs
    ):
        """Wywoływane tuż przed wywołaniem widoku."""
        logger.debug(f'Calling view: {view_func.__name__}')
        return None  # None oznacza kontynuację przetwarzania
    
    def process_exception(
        self, 
        request: HttpRequest, 
        exception: Exception
    ):
        """Obsługa wyjątków zgłoszonych przez widok."""
        if isinstance(exception, PermissionDenied):
            return JsonResponse(
                {'error': 'Access denied'},
                status=403
            )
        
        logger.error(
            f'Unhandled exception: {exception}\n'
            f'{traceback.format_exc()}'
        )
        return None  # Przekazanie do domyślnej obsługi
    
    def process_template_response(
        self, 
        request: HttpRequest, 
        response
    ):
        """Przetwarzanie odpowiedzi szablonowych."""
        if hasattr(response, 'context_data'):
            response.context_data['middleware_info'] = 'Processed'
        return response

Implementacja Zaawansowanego Systemu Logowania

Logowanie jest jednym z najczęstszych zastosowań middleware. Profesjonalna implementacja powinna uwzględniać strukturalne logowanie, korelację żądań oraz wydajność.

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 do strukturalnego logowania żądań 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:
        # Generowanie unikalnego ID żądania
        request_id = str(uuid.uuid4())
        request.request_id = request_id
        
        # Pomijanie logowania dla wykluczonych ścieżek
        if request.path in self.EXCLUDED_PATHS:
            return self.get_response(request)
        
        start_time = time.perf_counter()
        
        # Logowanie żądania
        self._log_request(request, request_id)
        
        try:
            response = self.get_response(request)
            duration = time.perf_counter() - start_time
            
            # Logowanie odpowiedzi
            self._log_response(request, response, request_id, duration)
            
            # Dodanie ID żądania do nagłówków odpowiedzi
            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:
        """Logowanie przychodzącego żądania."""
        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:
        """Logowanie odpowiedzi."""
        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:
        """Logowanie błędu."""
        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:
        """Pobranie adresu IP klienta z uwzględnieniem proxy."""
        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:
        """Pobranie nagłówków z ukryciem wrażliwych danych."""
        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 do Kontroli Dostępu i Rate Limiting

Kontrola dostępu i ograniczanie liczby żądań to krytyczne funkcjonalności dla aplikacji produkcyjnych.

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 ograniczający liczbę żądań na użytkownika/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):
        # Identyfikacja klienta
        client_id = self._get_client_identifier(request)
        cache_key = f'rate_limit:{client_id}'
        
        # Pobranie aktualnej liczby żądań
        request_data = cache.get(cache_key, {'count': 0, 'reset_at': 0})
        current_time = time.time()
        
        # Reset licznika jeśli okno czasowe wygasło
        if current_time > request_data['reset_at']:
            request_data = {
                'count': 0,
                'reset_at': current_time + self.rate_window
            }
        
        # Sprawdzenie limitu
        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)}
            )
        
        # Inkrementacja licznika
        request_data['count'] += 1
        cache.set(cache_key, request_data, self.rate_window)
        
        response = self.get_response(request)
        
        # Dodanie nagłówków informacyjnych
        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:
        """Identyfikacja klienta na podstawie użytkownika lub 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}'

Asynchroniczne Middleware w Django

Django w wersji 4.1 i nowszych wspiera asynchroniczne middleware, co jest szczególnie przydatne dla aplikacji wykorzystujących 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:
    """Asynchroniczne middleware integrujące się z zewnętrznym serwisem."""
    
    async_capable = True
    sync_capable = False
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    async def __call__(self, request: HttpRequest) -> HttpResponse:
        # Asynchroniczne pobranie danych z zewnętrznego serwisu
        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:
        """Asynchroniczne pobranie danych z zewnętrznego 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 obsługujące zarówno sync jak i async."""
    
    async_capable = True
    sync_capable = True
    
    def __init__(self, get_response):
        self.get_response = get_response
        # Wykrycie czy get_response jest 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)

Testowanie Middleware

Testowanie middleware wymaga szczególnego podejścia ze względu na ich pozycję w łańcuchu przetwarzania żądań.

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):
    """Testy dla 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):
        """Test czy middleware dodaje nagłówek z czasem wykonania."""
        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):
        """Test czy middleware wywołuje następny element łańcucha."""
        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):
        """Test czy middleware loguje informacje o żądaniu."""
        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):
    """Testy dla 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):
        """Test czy middleware przepuszcza żądania w limicie."""
        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):
        """Test czy middleware blokuje żądania przekraczające limit."""
        middleware = RateLimitMiddleware(self.get_response)
        request = self.factory.get('/test/')
        request.user = MagicMock(is_authenticated=False)
        request.META['REMOTE_ADDR'] = '192.168.1.2'
        
        # Dwa pierwsze żądania powinny przejść
        middleware(request)
        middleware(request)
        
        # Trzecie żądanie powinno być zablokowane
        response = middleware(request)
        self.assertEqual(response.status_code, 429)

Gotowy na rozmowy o Django?

Ćwicz z naszymi interaktywnymi symulatorami, flashcards i testami technicznymi.

Odkryj Django

Pytania Rekrutacyjne dotyczące Middleware w Django

Podczas rozmów kwalifikacyjnych na stanowiska związane z Django często pojawiają się pytania dotyczące middleware. Oto najważniejsze zagadnienia wraz z odpowiedziami.

Pytanie: Jaka jest różnica między MIDDLEWARE a MIDDLEWARE_CLASSES w Django?

MIDDLEWARE_CLASSES to stara konfiguracja używana w Django < 1.10, która wymagała dziedziczenia po MiddlewareMixin. MIDDLEWARE to nowy styl wprowadzony w Django 1.10, wykorzystujący prostszy wzorzec z metodą __call__ i parametrem get_response. Nowy styl jest bardziej elastyczny i wspiera zarówno synchroniczne jak i asynchroniczne middleware.

Pytanie: W jakiej kolejności wykonywane są middleware podczas przetwarzania żądania i odpowiedzi?

Middleware są wykonywane w kolejności zdefiniowanej w MIDDLEWARE podczas przetwarzania żądania (od góry do dołu), a następnie w odwrotnej kolejności podczas przetwarzania odpowiedzi (od dołu do góry). Ta architektura przypomina cebulę lub stos LIFO.

Pytanie: Jak przerwać łańcuch middleware i zwrócić odpowiedź wcześniej?

Aby przerwać łańcuch, middleware powinno zwrócić obiekt HttpResponse bezpośrednio w metodzie __call__ zamiast wywoływać get_response. Przykładem jest middleware sprawdzające autoryzację, które zwraca odpowiedź 403 bez dalszego przetwarzania.

Pytanie: Kiedy używa się process_view a kiedy process_exception?

process_view jest wywoływane tuż przed wywołaniem widoku i ma dostęp do funkcji widoku oraz jej argumentów. Może służyć do kontroli dostępu lub modyfikacji argumentów. process_exception jest wywoływane tylko gdy widok zgłosi wyjątek i służy do centralnej obsługi błędów.

Pytanie: Jak zaimplementować middleware obsługujące zarówno WSGI jak i ASGI?

Należy ustawić atrybuty klasy sync_capable = True i async_capable = True, a następnie wykryć w konstruktorze czy get_response jest funkcją asynchroniczną. W metodzie __call__ należy odpowiednio wywołać wersję sync lub async przetwarzania.

Najlepsze Praktyki i Wzorce

Przy implementacji middleware warto przestrzegać kilku zasad:

  1. Zasada pojedynczej odpowiedzialności — każde middleware powinno realizować jedną, jasno określoną funkcję
  2. Wydajność — middleware jest wywoływane dla każdego żądania, więc operacje powinny być maksymalnie zoptymalizowane
  3. Odporność na błędy — middleware nie powinno powodować awarii aplikacji; należy odpowiednio obsługiwać wyjątki
  4. Konfigurowalność — parametry middleware powinny być konfigurowalne przez settings.py
  5. Testowalność — middleware powinno być łatwe do testowania w izolacji

Podsumowanie

Middleware w Django to potężne narzędzie pozwalające na implementację przekrojowych funkcjonalności w sposób modularny i reużywalny. Zrozumienie architektury middleware, umiejętność tworzenia własnych implementacji oraz znajomość najlepszych praktyk są niezbędne dla każdego zaawansowanego programisty Django. Przedstawione w artykule przykłady obejmują najczęstsze przypadki użycia: logowanie, kontrolę dostępu, rate limiting oraz obsługę asynchroniczną, stanowiąc solidną bazę do dalszego rozwijania własnych rozwiązań.

Wyzwanie dnia

Znajdziesz błąd w Django?

Prawdziwy fragment kodu, ukryty błąd, jedna próba dziennie. Bez konta, żeby spróbować.

Podejmij wyzwanie
Anthony Fillion-Maillet

Autor:

Anthony Fillion-Maillet

Założyciel SharpSkill

Programista fullstack od ponad 10 lat. Prowadzi SharpSkill i odpowiada za wszystko, co się tu ukazuje.

Zaktualizowano 18 września 2026

Tagi

#django
#middleware
#python
#logowanie
#backend

Udostępnij

Powiązane artykuły

Django Signals vs Celery Tasks 2026

Django Signals vs Celery Tasks w 2026: Kiedy Stosować i Pytania Rekrutacyjne

Kompleksowe porównanie Django Signals i Celery Tasks w 2026 roku. Kiedy używać sygnałów, a kiedy zadań asynchronicznych, z przykładami kodu i pytaniami na rozmowę kwalifikacyjną.

Diagram architektury middleware i sygnalow w Django 5.2 z przeplywem requestow i zdarzeniami domenowymi

Django 5.2: Wlasne Middleware i Obsluga Sygnalow -- Przygotowanie do Rozmow Rekrutacyjnych

Praktyczny przewodnik po tworzeniu wlasnych middleware i obsludze sygnalow w Django 5.2. Struktura middleware, asynchroniczne middleware, sygnaly post_save i pre_save, wlasne sygnaly domenowe oraz najczesciej zadawane pytania rekrutacyjne.

Przygotowanie do rozmowy kwalifikacyjnej z Django obejmujące ORM, Middleware i Django REST Framework

Pytania rekrutacyjne Django: ORM, Middleware i DRF, szczegółowa analiza

Pytania rekrutacyjne Django obejmujące optymalizację ORM z select_related, prefetch_related i trybem FETCH_PEERS w Django 6.1, architekturę middleware oraz wydajność serializerów Django REST Framework, uprawnienia i paginację.