2026'da Django'da İleri Düzey Middleware: Özel Middleware Oluşturma, Loglama ve Mülakat Soruları

Django 2026'da ileri düzey middleware tekniklerini kapsayan detaylı rehber. Özel middleware oluşturma, loglama implementasyonu ve teknik mülakat sorularına hazırlık.

2026'da Django'da ileri düzey middleware

Django middleware, HTTP istekleri ile view'lar arasında köprü görevi gören çerçevenin en kritik bileşenlerinden birini temsil eder. 2026 yılında middleware, Django geliştiricileri için view mantığını değiştirmeden kesişen işlevlerin uygulanmasını sağlayan temel bir araç olmaya devam etmektedir.

Django middleware soğan modeli gibi çalışır — her katman isteği view'a giderken işler, ardından yanıtı istemciye dönerken tekrar işler. Bu akışı anlamak, middleware'i etkili kullanmak için kritik öneme sahiptir.

Django'da Middleware Mimarisi

Django middleware, uygulama üzerinden geçen her HTTP isteği için çağrılan sınıflar veya fonksiyonlardır. Middleware sistemi, her middleware'in şunları yapabildiği sorumluluk zinciri desenine göre çalışır:

  • İsteği bir sonraki middleware veya view'a iletmeden önce işleme
  • Yanıtı istemciye döndürmeden önce işleme
  • Zinciri kesip kendi yanıtını döndürme
  • View'lar veya diğer middleware'ler tarafından fırlatılan istisnaları işleme
python
# settings.py - Middleware sıralaması önemlidir
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',
    # Özel middleware genellikle sona eklenir
    'myapp.middleware.CustomMiddleware',
]

Sınıf Tabanlı Özel Middleware Oluşturma

Django, middleware tanımlamak için iki stil destekler: sınıf tabanlı ve fonksiyon tabanlı. Sınıf tabanlı stil daha fazla esneklik sunar ve karmaşık kullanım durumları için tercih edilir.

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:
    """İstek yürütme süresini ölçen middleware."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        # Sunucu başlangıcında çalışan tek seferlik yapılandırma
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # View'dan önce çalışan kod
        start_time = time.perf_counter()
        
        # Sonraki middleware veya view'ı çağır
        response = self.get_response(request)
        
        # View'dan sonra çalışan kod
        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

View ve İstisna İşleme Hook'larıyla Middleware

İleri düzey senaryolar için Django, istek işlemenin farklı aşamaları üzerinde hassas kontrol sağlayan özel hook metodları sunar.

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:
    """Tam hook seti içeren ileri düzey middleware."""
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    def __call__(self, request: HttpRequest) -> HttpResponse:
        # İstek ön işleme
        self.process_request(request)
        
        response = self.get_response(request)
        
        # Yanıt son işleme
        return self.process_response(request, response)
    
    def process_request(self, request: HttpRequest) -> None:
        """View'a iletmeden önce işleme."""
        request.custom_data = {
            'middleware_processed': True,
            'timestamp': time.time()
        }
    
    def process_response(
        self, 
        request: HttpRequest, 
        response: HttpResponse
    ) -> HttpResponse:
        """View yürütüldükten sonra işleme."""
        response['X-Processed-By'] = 'AdvancedMiddleware'
        return response
    
    def process_view(
        self, 
        request: HttpRequest, 
        view_func, 
        view_args, 
        view_kwargs
    ):
        """View çağrılmadan hemen önce çağrılır."""
        logger.debug(f'Calling view: {view_func.__name__}')
        return None  # None işlemeye devam anlamına gelir
    
    def process_exception(
        self, 
        request: HttpRequest, 
        exception: Exception
    ):
        """View tarafından fırlatılan istisnaları işleme."""
        if isinstance(exception, PermissionDenied):
            return JsonResponse(
                {'error': 'Access denied'},
                status=403
            )
        
        logger.error(
            f'Unhandled exception: {exception}\n'
            f'{traceback.format_exc()}'
        )
        return None  # Varsayılan işlemeye ilet
    
    def process_template_response(
        self, 
        request: HttpRequest, 
        response
    ):
        """Şablon yanıtlarını işleme."""
        if hasattr(response, 'context_data'):
            response.context_data['middleware_info'] = 'Processed'
        return response

İleri Düzey Loglama Sistemi Implementasyonu

Loglama, middleware'in en yaygın kullanım alanlarından biridir. Profesyonel bir implementasyon yapısal loglama, istek korelasyonu ve performansı dikkate almalıdır.

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:
    """HTTP istekleri için yapısal loglama middleware'i."""
    
    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:
        # Benzersiz istek ID'si oluşturma
        request_id = str(uuid.uuid4())
        request.request_id = request_id
        
        # Hariç tutulan yollar için loglamayı atlama
        if request.path in self.EXCLUDED_PATHS:
            return self.get_response(request)
        
        start_time = time.perf_counter()
        
        # İsteği loglama
        self._log_request(request, request_id)
        
        try:
            response = self.get_response(request)
            duration = time.perf_counter() - start_time
            
            # Yanıtı loglama
            self._log_response(request, response, request_id, duration)
            
            # Yanıt başlıklarına istek ID'si ekleme
            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:
        """Gelen isteği loglama."""
        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:
        """Yanıtı loglama."""
        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:
        """Hatayı loglama."""
        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:
        """Proxy'yi dikkate alarak istemci IP adresini alma."""
        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:
        """Hassas verileri gizleyerek başlıkları alma."""
        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

Erişim Kontrolü ve Rate Limiting Middleware

Erişim kontrolü ve istek sınırlandırma, üretim uygulamaları için kritik işlevlerdir.

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:
    """Kullanıcı/IP başına istek sayısını sınırlayan middleware."""
    
    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):
        # İstemci tanımlama
        client_id = self._get_client_identifier(request)
        cache_key = f'rate_limit:{client_id}'
        
        # Mevcut istek sayısını alma
        request_data = cache.get(cache_key, {'count': 0, 'reset_at': 0})
        current_time = time.time()
        
        # Zaman penceresi sona erdiyse sayacı sıfırla
        if current_time > request_data['reset_at']:
            request_data = {
                'count': 0,
                'reset_at': current_time + self.rate_window
            }
        
        # Limit kontrolü
        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)}
            )
        
        # Sayacı artırma
        request_data['count'] += 1
        cache.set(cache_key, request_data, self.rate_window)
        
        response = self.get_response(request)
        
        # Bilgilendirme başlıkları ekleme
        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:
        """Kullanıcı veya IP'ye göre istemci tanımlama."""
        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}'

Django'da Asenkron Middleware

Django 4.1 ve sonraki sürümler, özellikle ASGI kullanan uygulamalar için yararlı olan asenkron middleware'i destekler.

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

logger = logging.getLogger(__name__)


class AsyncExternalServiceMiddleware:
    """Harici servislerle entegre olan asenkron middleware."""
    
    async_capable = True
    sync_capable = False
    
    def __init__(self, get_response):
        self.get_response = get_response
        
    async def __call__(self, request: HttpRequest) -> HttpResponse:
        # Harici servisten asenkron veri alma
        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:
        """Harici API'den asenkron veri alma."""
        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:
    """Hem sync hem async destekleyen middleware."""
    
    async_capable = True
    sync_capable = True
    
    def __init__(self, get_response):
        self.get_response = get_response
        # get_response'un async olup olmadığını tespit etme
        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 Testi

Middleware testi, istek işleme zincirindeki konumu nedeniyle özel bir yaklaşım gerektirir.

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 için testler."""
    
    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'in süre başlığı ekleyip eklemediğini test eder."""
        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'in zincirdeki sonraki öğeyi çağırıp çağırmadığını test eder."""
        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'in istek bilgilerini loglayıp loglamadığını test eder."""
        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 için testler."""
    
    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'in limit içindeki isteklere izin verip vermediğini test eder."""
        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'in limiti aşan istekleri engelleyip engellemediğini test eder."""
        middleware = RateLimitMiddleware(self.get_response)
        request = self.factory.get('/test/')
        request.user = MagicMock(is_authenticated=False)
        request.META['REMOTE_ADDR'] = '192.168.1.2'
        
        # İlk iki istek geçmeli
        middleware(request)
        middleware(request)
        
        # Üçüncü istek engellenmeli
        response = middleware(request)
        self.assertEqual(response.status_code, 429)

Django mülakatlarında başarılı olmaya hazır mısın?

İnteraktif simülatörler, flashcards ve teknik testlerle pratik yap.

Django Middleware Mülakat Soruları

Django pozisyonları için yapılan teknik mülakatlarda middleware konusu sıkça sorulmaktadır. İşte en önemli konular ve cevapları.

Soru: Django'da MIDDLEWARE ile MIDDLEWARE_CLASSES arasındaki fark nedir?

MIDDLEWARE_CLASSES, Django < 1.10'da kullanılan ve MiddlewareMixin'den kalıtım gerektiren eski yapılandırmadır. MIDDLEWARE, Django 1.10'da tanıtılan ve __call__ metodu ile get_response parametresini kullanan daha basit bir desendir. Yeni stil daha esnektir ve hem senkron hem de asenkron middleware'i destekler.

Soru: İstek ve yanıt işleme sırasında middleware hangi sırada çalışır?

Middleware'ler, istek işleme sırasında MIDDLEWARE'de tanımlanan sırada (yukarıdan aşağıya), yanıt işleme sırasında ise ters sırada (aşağıdan yukarıya) çalışır. Bu mimari soğan veya LIFO yığını gibi düşünülebilir.

Soru: Middleware zinciri nasıl kesilir ve erken yanıt döndürülür?

Zinciri kesmek için middleware, get_response'u çağırmak yerine doğrudan __call__ metodunda bir HttpResponse nesnesi döndürmelidir. Örneğin, yetkilendirme kontrolü yapan bir middleware, daha fazla işlem yapmadan 403 yanıtı döndürebilir.

Soru: process_view ve process_exception ne zaman kullanılır?

process_view, view çağrılmadan hemen önce çağrılır ve view fonksiyonuna ve argümanlarına erişim sağlar. Erişim kontrolü veya argüman değişikliği için kullanılabilir. process_exception yalnızca view bir istisna fırlattığında çağrılır ve merkezi hata işleme için kullanılır.

Soru: Hem WSGI hem ASGI'yi destekleyen middleware nasıl implement edilir?

Sınıf öznitelikleri olarak sync_capable = True ve async_capable = True ayarlanmalı, ardından constructor'da get_response'un asenkron bir fonksiyon olup olmadığı tespit edilmelidir. __call__ metodunda işlemenin sync veya async versiyonu uygun şekilde çağrılmalıdır.

En İyi Uygulamalar ve Desenler

Middleware implementasyonunda birkaç ilkeye uymak önemlidir:

  1. Tek Sorumluluk İlkesi — Her middleware açıkça tanımlanmış tek bir işlevi yerine getirmelidir
  2. Performans — Middleware her istek için çağrıldığından işlemler maksimum düzeyde optimize edilmelidir
  3. Hata Dayanıklılığı — Middleware uygulama çökmesine neden olmamalı; istisnalar düzgün işlenmelidir
  4. Yapılandırılabilirlik — Middleware parametreleri settings.py üzerinden yapılandırılabilir olmalıdır
  5. Test Edilebilirlik — Middleware izole olarak kolayca test edilebilir olmalıdır

Sonuç

Django middleware, kesişen işlevlerin modüler ve yeniden kullanılabilir şekilde uygulanmasını sağlayan güçlü bir araçtır. Middleware mimarisini anlamak, özel implementasyonlar oluşturabilmek ve en iyi uygulamaları bilmek, her ileri düzey Django geliştiricisi için gereklidir. Bu makalede sunulan örnekler en yaygın kullanım durumlarını kapsar: loglama, erişim kontrolü, rate limiting ve asenkron işleme, kendi çözümlerini geliştirmek için sağlam bir temel oluşturur.

Günün meydan okuması

Django kodundaki hatayı bulabilir misin?

Gerçek bir kod parçası, gizli bir hata, günde bir deneme. Denemek için hesap gerekmez.

Anthony Fillion-Maillet

Yazan:

Anthony Fillion-Maillet

SharpSkill kurucusu

10 yılı aşkın süredir fullstack geliştirici. SharpSkill’i yönetiyor ve burada yayımlanan her şeyden sorumlu.

18 Eylül 2026 tarihinde güncellendi

Etiketler

#django
#middleware
#python
#loglama
#backend

Paylaş

İlgili makaleler