Django Middleware Deep Dive: Build Custom Middleware Like a Pro

Middleware is one of Django's most powerful features, acting as a framework of hooks into Django's request/response processing. Understanding middleware is essential for building robust, scalable Django applications.

What is Middleware?

Middleware sits between the client and your view, processing every request and response that flows through your application.

┌─────────────────────────────────────────────────────────────────┐
│                        Django Application                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   Request Flow:                                                  │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌─────────┐  │
│   │  Client  │───▶│Middleware│───▶│Middleware│───▶│  View   │  │
│   │          │    │    1     │    │    2     │    │         │  │
│   └──────────┘    └──────────┘    └──────────┘    └─────────┘  │
│                                                                  │
│   Response Flow:                                                 │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌─────────┐  │
│   │  Client  │◀───│Middleware│◀───│Middleware│◀───│  View   │  │
│   │          │    │    1     │    │    2     │    │         │  │
│   └──────────┘    └──────────┘    └──────────┘    └─────────┘  │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Middleware Execution Order

Django processes middleware in the order they're defined in MIDDLEWARE:

# settings.py
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',           # 1st request, last response
    'django.contrib.sessions.middleware.SessionMiddleware',    # 2nd request, 2nd-last response
    'django.middleware.common.CommonMiddleware',               # 3rd request, 3rd-last response
    'django.middleware.csrf.CsrfViewMiddleware',              # ... and so on
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

Understanding the Middleware Hooks

Django middleware has several hooks for different stages:

class MyMiddleware:
    def __init__(self, get_response):
        """
        One-time configuration and initialization.
        Called once when the server starts.
        """
        self.get_response = get_response

    def __call__(self, request):
        """
        Code executed for each request BEFORE the view.
        """
        # Pre-processing logic here

        response = self.get_response(request)

        # Post-processing logic here
        # Code executed for each request AFTER the view

        return response

    def process_view(self, request, view_func, view_args, view_kwargs):
        """
        Called just before Django calls the view.
        Return None to continue processing, or HttpResponse to short-circuit.
        """
        pass

    def process_exception(self, request, exception):
        """
        Called when a view raises an exception.
        Return HttpResponse to handle, or None to let Django handle it.
        """
        pass

    def process_template_response(self, request, response):
        """
        Called after the view if response has a render() method.
        Must return a response with render() method.
        """
        return response

Practical Middleware Examples

1. Request Logging Middleware

Track all incoming requests for debugging and analytics:

# middleware/logging_middleware.py
import logging
import time
import json
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger('request_logger')

class RequestLoggingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Start timer
        start_time = time.time()

        # Get request body for POST/PUT/PATCH
        request_body = None
        if request.method in ['POST', 'PUT', 'PATCH']:
            try:
                request_body = request.body.decode('utf-8')[:1000]  # Limit size
            except:
                request_body = '<binary data>'

        # Process request
        response = self.get_response(request)

        # Calculate duration
        duration = time.time() - start_time

        # Log request details
        log_data = {
            'method': request.method,
            'path': request.path,
            'status_code': response.status_code,
            'duration_ms': round(duration * 1000, 2),
            'user': str(request.user) if hasattr(request, 'user') else 'Anonymous',
            'ip': self.get_client_ip(request),
            'user_agent': request.META.get('HTTP_USER_AGENT', ''),
        }

        if response.status_code >= 400:
            logger.warning(json.dumps(log_data))
        else:
            logger.info(json.dumps(log_data))

        return response

    def get_client_ip(self, request):
        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')

2. Rate Limiting Middleware

Protect your API from abuse:

# middleware/rate_limit_middleware.py
from django.core.cache import cache
from django.http import JsonResponse
import time

class RateLimitMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        # Configuration
        self.rate_limit = 100  # requests
        self.time_window = 60  # seconds

    def __call__(self, request):
        # Skip rate limiting for certain paths
        if request.path.startswith('/admin/'):
            return self.get_response(request)

        # Get client identifier
        client_id = self.get_client_id(request)
        cache_key = f'rate_limit:{client_id}'

        # Get current request count
        request_data = cache.get(cache_key, {'count': 0, 'start_time': time.time()})

        # Reset if time window has passed
        if time.time() - request_data['start_time'] > self.time_window:
            request_data = {'count': 0, 'start_time': time.time()}

        # Increment count
        request_data['count'] += 1

        # Check if rate limit exceeded
        if request_data['count'] > self.rate_limit:
            retry_after = int(self.time_window - (time.time() - request_data['start_time']))
            return JsonResponse({
                'error': 'Rate limit exceeded',
                'retry_after': retry_after,
                'limit': self.rate_limit,
                'window': self.time_window
            }, status=429, headers={'Retry-After': str(retry_after)})

        # Update cache
        cache.set(cache_key, request_data, self.time_window)

        # Add rate limit headers to response
        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['start_time'] + self.time_window))

        return response

    def get_client_id(self, request):
        # Use user ID if authenticated, otherwise use IP
        if hasattr(request, 'user') and request.user.is_authenticated:
            return f'user:{request.user.id}'

        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return f'ip:{x_forwarded_for.split(",")[0].strip()}'
        return f'ip:{request.META.get("REMOTE_ADDR")}'

3. JWT Authentication Middleware

Handle token-based authentication:

# middleware/jwt_middleware.py
import jwt
from django.conf import settings
from django.http import JsonResponse
from django.contrib.auth import get_user_model

User = get_user_model()

class JWTAuthenticationMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.exempt_paths = [
            '/api/auth/login/',
            '/api/auth/register/',
            '/api/auth/refresh/',
            '/health/',
        ]

    def __call__(self, request):
        # Skip authentication for exempt paths
        if any(request.path.startswith(path) for path in self.exempt_paths):
            return self.get_response(request)

        # Skip for non-API routes
        if not request.path.startswith('/api/'):
            return self.get_response(request)

        # Get token from header
        auth_header = request.META.get('HTTP_AUTHORIZATION', '')

        if not auth_header.startswith('Bearer '):
            return JsonResponse({
                'error': 'Authorization header missing or invalid',
                'detail': 'Expected: Bearer <token>'
            }, status=401)

        token = auth_header.split(' ')[1]

        try:
            # Decode and verify token
            payload = jwt.decode(
                token,
                settings.SECRET_KEY,
                algorithms=['HS256']
            )

            # Get user from token
            user = User.objects.get(id=payload['user_id'])

            if not user.is_active:
                return JsonResponse({
                    'error': 'User account is disabled'
                }, status=401)

            # Attach user to request
            request.user = user
            request.jwt_payload = payload

        except jwt.ExpiredSignatureError:
            return JsonResponse({
                'error': 'Token has expired',
                'code': 'TOKEN_EXPIRED'
            }, status=401)

        except jwt.InvalidTokenError as e:
            return JsonResponse({
                'error': 'Invalid token',
                'detail': str(e)
            }, status=401)

        except User.DoesNotExist:
            return JsonResponse({
                'error': 'User not found'
            }, status=401)

        return self.get_response(request)

4. Request/Response Transformation Middleware

Automatically handle JSON responses and add metadata:

# middleware/api_middleware.py
import json
import time
from django.http import JsonResponse

class APIResponseMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Only process API routes
        if not request.path.startswith('/api/'):
            return self.get_response(request)

        # Add request timestamp
        request.start_time = time.time()

        response = self.get_response(request)

        # Transform JSON responses
        if hasattr(response, 'content') and response.get('Content-Type', '').startswith('application/json'):
            try:
                data = json.loads(response.content)

                # Wrap response in standard format
                wrapped_response = {
                    'success': 200 <= response.status_code < 400,
                    'status_code': response.status_code,
                    'data': data if response.status_code < 400 else None,
                    'error': data if response.status_code >= 400 else None,
                    'meta': {
                        'request_id': request.META.get('HTTP_X_REQUEST_ID', ''),
                        'timestamp': time.time(),
                        'duration_ms': round((time.time() - request.start_time) * 1000, 2),
                        'version': 'v1'
                    }
                }

                response.content = json.dumps(wrapped_response)

            except json.JSONDecodeError:
                pass

        return response

5. CORS Middleware (Simple Implementation)

Handle Cross-Origin Resource Sharing:

# middleware/cors_middleware.py
from django.conf import settings

class CORSMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.allowed_origins = getattr(settings, 'CORS_ALLOWED_ORIGINS', [])
        self.allow_all = getattr(settings, 'CORS_ALLOW_ALL_ORIGINS', False)

    def __call__(self, request):
        # Handle preflight requests
        if request.method == 'OPTIONS':
            response = self.get_response(request)
            return self.add_cors_headers(request, response)

        response = self.get_response(request)
        return self.add_cors_headers(request, response)

    def add_cors_headers(self, request, response):
        origin = request.META.get('HTTP_ORIGIN', '')

        # Check if origin is allowed
        if self.allow_all:
            response['Access-Control-Allow-Origin'] = '*'
        elif origin in self.allowed_origins:
            response['Access-Control-Allow-Origin'] = origin
            response['Vary'] = 'Origin'

        # Add other CORS headers
        response['Access-Control-Allow-Methods'] = 'GET, POST, PUT, PATCH, DELETE, OPTIONS'
        response['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-Requested-With'
        response['Access-Control-Max-Age'] = '86400'  # 24 hours
        response['Access-Control-Allow-Credentials'] = 'true'

        return response

Middleware Best Practices

1. Order Matters

# settings.py - Correct order
MIDDLEWARE = [
    # Security first
    'django.middleware.security.SecurityMiddleware',

    # Session before auth
    'django.contrib.sessions.middleware.SessionMiddleware',

    # CORS before CommonMiddleware (for preflight)
    'myapp.middleware.CORSMiddleware',

    'django.middleware.common.CommonMiddleware',

    # CSRF after session
    'django.middleware.csrf.CsrfViewMiddleware',

    # Auth after session and CSRF
    'django.contrib.auth.middleware.AuthenticationMiddleware',

    # Custom middleware after Django's core
    'myapp.middleware.RequestLoggingMiddleware',
    'myapp.middleware.RateLimitMiddleware',
]

2. Handle Exceptions Gracefully

class SafeMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        try:
            # Your middleware logic
            self.before_request(request)
        except Exception as e:
            # Log but don't break the request
            logger.exception(f"Middleware error: {e}")

        response = self.get_response(request)

        try:
            self.after_response(request, response)
        except Exception as e:
            logger.exception(f"Middleware error: {e}")

        return response

3. Use Caching Wisely

from django.core.cache import cache
from functools import lru_cache

class CachedConfigMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    @lru_cache(maxsize=1)
    def get_config(self):
        # Cache configuration that doesn't change often
        return cache.get('app_config', default_config)

    def __call__(self, request):
        request.app_config = self.get_config()
        return self.get_response(request)

4. Make Middleware Configurable

# middleware/configurable_middleware.py
from django.conf import settings

class ConfigurableMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        # Load configuration from settings
        self.config = getattr(settings, 'MY_MIDDLEWARE_CONFIG', {})
        self.enabled = self.config.get('enabled', True)
        self.exempt_paths = self.config.get('exempt_paths', [])

    def __call__(self, request):
        if not self.enabled:
            return self.get_response(request)

        if any(request.path.startswith(p) for p in self.exempt_paths):
            return self.get_response(request)

        # Middleware logic here
        return self.get_response(request)

Testing Middleware

# tests/test_middleware.py
from django.test import TestCase, RequestFactory, override_settings
from django.contrib.auth import get_user_model
from myapp.middleware import RateLimitMiddleware

User = get_user_model()

class RateLimitMiddlewareTest(TestCase):
    def setUp(self):
        self.factory = RequestFactory()
        self.middleware = RateLimitMiddleware(lambda r: HttpResponse('OK'))

    def test_allows_requests_under_limit(self):
        request = self.factory.get('/api/test/')
        response = self.middleware(request)
        self.assertEqual(response.status_code, 200)

    def test_blocks_requests_over_limit(self):
        request = self.factory.get('/api/test/')

        # Make requests up to limit
        for i in range(101):
            response = self.middleware(request)

        self.assertEqual(response.status_code, 429)
        self.assertIn('rate limit exceeded', response.content.decode().lower())

    def test_rate_limit_headers_present(self):
        request = self.factory.get('/api/test/')
        response = self.middleware(request)

        self.assertIn('X-RateLimit-Limit', response)
        self.assertIn('X-RateLimit-Remaining', response)

Conclusion

Middleware is essential for building production-ready Django applications. Key takeaways:

1. Understand the execution order - middleware processes requests in order, responses in reverse
2. Use appropriate hooks - __call__, process_view, process_exception, process_template_response
3. Handle exceptions gracefully - don't let middleware errors break your application
4. Keep middleware focused - each middleware should do one thing well
5. Test thoroughly - middleware affects every request, so test edge cases

With these patterns, you can build robust middleware for logging, authentication, rate limiting, and more.

Further Reading

• Django Middleware Documentation