Django Middleware — Deep Dive
How Django builds the middleware chain
When Django starts, it reads MIDDLEWARE from settings and constructs a handler chain. The process works inside out: the last middleware wraps the view, the second-to-last wraps that, and so on until the first middleware wraps everything.
# Simplified from django.core.handlers.base.BaseHandler
def load_middleware(self):
handler = convert_exception_to_response(self._get_response)
for middleware_path in reversed(settings.MIDDLEWARE):
middleware = import_string(middleware_path)
handler = middleware(handler)
self._middleware_chain = handler
Each middleware’s __init__ receives the next handler in the chain as get_response. When a request arrives, Django calls self._middleware_chain(request), which cascades through the nested handlers.
This design means __init__ runs once at startup (not per request), making it the right place for one-time configuration like compiling regex patterns or loading configuration.
Synchronous vs async middleware
Django 4.1+ supports async middleware natively. The framework inspects your middleware class to determine its mode:
import asyncio
class AsyncTimingMiddleware:
async_capable = True
sync_capable = False
def __init__(self, get_response):
self.get_response = get_response
if asyncio.iscoroutinefunction(self.get_response):
markcoroutinefunction(self)
async def __call__(self, request):
import time
start = time.monotonic()
response = await self.get_response(request)
duration = time.monotonic() - start
response['X-Request-Duration'] = f'{duration:.4f}'
return response
If your middleware is sync-only but the view is async, Django wraps the middleware in sync_to_async. This adds thread-pool overhead. For maximum performance in async deployments, mark middleware as async_capable = True and implement both sync and async paths:
from asgiref.sync import iscoroutinefunction, markcoroutinefunction
class DualModeMiddleware:
async_capable = True
sync_capable = True
def __init__(self, get_response):
self.get_response = get_response
if iscoroutinefunction(self.get_response):
markcoroutinefunction(self)
def __call__(self, request):
if iscoroutinefunction(self):
return self.__acall__(request)
# Sync path
self.before_request(request)
response = self.get_response(request)
self.after_response(request, response)
return response
async def __acall__(self, request):
self.before_request(request)
response = await self.get_response(request)
self.after_response(request, response)
return response
Exception handling pipeline
When a view raises an exception, Django’s middleware exception handling follows a specific path:
- The exception propagates up through the middleware chain
- Each middleware’s
process_exceptionis called in reverse order - If any
process_exceptionreturns a response, that response continues through the remaining response-phase middleware - If no middleware handles the exception, Django’s default exception handler produces a 500 response
class DetailedErrorMiddleware:
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):
if isinstance(exception, PermissionError):
return HttpResponseForbidden(
json.dumps({
'error': 'permission_denied',
'detail': str(exception),
'request_id': getattr(request, 'request_id', None)
}),
content_type='application/json'
)
# Return None to let other middleware or Django handle it
return None
Important: process_exception doesn’t catch exceptions in other middleware — only in the view and in middleware below it in the stack.
Request-scoped state with middleware
Attaching state to the request object is a common middleware pattern for passing data to views without function parameters:
import uuid
class RequestIDMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
request.request_id = request.META.get(
'HTTP_X_REQUEST_ID',
str(uuid.uuid4())
)
response = self.get_response(request)
response['X-Request-ID'] = request.request_id
return response
For tracing, propagate this ID into logging context using a filter:
import logging
import threading
_request_id = threading.local()
class RequestIDFilter(logging.Filter):
def filter(self, record):
record.request_id = getattr(_request_id, 'value', 'no-request')
return True
class RequestIDMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
request.request_id = request.META.get(
'HTTP_X_REQUEST_ID', str(uuid.uuid4())
)
_request_id.value = request.request_id
try:
response = self.get_response(request)
finally:
_request_id.value = None
response['X-Request-ID'] = request.request_id
return response
Middleware for database connection management
In multi-database setups, middleware can route read queries to replicas:
import random
class ReadReplicaMiddleware:
def __init__(self, get_response):
self.get_response = get_response
self.replicas = ['replica1', 'replica2', 'replica3']
def __call__(self, request):
if request.method in ('GET', 'HEAD', 'OPTIONS'):
request._db_alias = random.choice(self.replicas)
else:
request._db_alias = 'default'
return self.get_response(request)
Pair this with a database router that reads request._db_alias (accessible via thread-local storage or context variables).
Security middleware patterns
IP-based access control
from ipaddress import ip_address, ip_network
class IPAllowlistMiddleware:
def __init__(self, get_response):
self.get_response = get_response
self.allowed_networks = [
ip_network(cidr) for cidr in settings.ALLOWED_IP_RANGES
]
def __call__(self, request):
client_ip = ip_address(self.get_client_ip(request))
if not any(client_ip in net for net in self.allowed_networks):
return HttpResponseForbidden('Access denied')
return self.get_response(request)
def get_client_ip(self, request):
forwarded = request.META.get('HTTP_X_FORWARDED_FOR')
if forwarded:
return forwarded.split(',')[0].strip()
return request.META.get('REMOTE_ADDR')
Response header hardening
class SecurityHeadersMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
response['X-Content-Type-Options'] = 'nosniff'
response['X-Frame-Options'] = 'DENY'
response['Referrer-Policy'] = 'strict-origin-when-cross-origin'
response['Permissions-Policy'] = 'geolocation=(), camera=()'
if request.is_secure():
response['Strict-Transport-Security'] = (
'max-age=31536000; includeSubDomains; preload'
)
return response
Performance profiling middleware
For production profiling, conditionally enable query logging:
from django.db import connection, reset_queries
import json
import logging
logger = logging.getLogger('performance')
class QueryProfileMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
if not request.META.get('HTTP_X_PROFILE_QUERIES'):
return self.get_response(request)
reset_queries()
response = self.get_response(request)
queries = connection.queries
total_time = sum(float(q['time']) for q in queries)
logger.info(
'Query profile for %s: %d queries, %.3fs total',
request.path, len(queries), total_time
)
if request.META.get('HTTP_X_PROFILE_DETAIL'):
response['X-Query-Count'] = str(len(queries))
response['X-Query-Time'] = f'{total_time:.3f}'
return response
Testing middleware in isolation
Test middleware without running the full Django request cycle:
from django.test import RequestFactory, TestCase
class TestTimingMiddleware(TestCase):
def setUp(self):
self.factory = RequestFactory()
def test_adds_duration_header(self):
def mock_view(request):
return HttpResponse('OK')
middleware = TimingMiddleware(mock_view)
request = self.factory.get('/test/')
response = middleware(request)
self.assertIn('X-Request-Duration', response)
duration = float(response['X-Request-Duration'])
self.assertGreater(duration, 0)
def test_short_circuits_on_blocked_ip(self):
def mock_view(request):
return HttpResponse('OK')
middleware = IPAllowlistMiddleware(mock_view)
request = self.factory.get('/test/')
request.META['REMOTE_ADDR'] = '1.2.3.4'
response = middleware(request)
self.assertEqual(response.status_code, 403)
Using RequestFactory keeps tests fast and isolated from URL routing and template rendering.
Tradeoffs and architecture decisions
Middleware is powerful but comes with costs. Every middleware adds latency to every request. Even a middleware that does nothing but call get_response adds function call overhead multiplied by request volume.
For high-traffic sites, audit your middleware stack regularly. Remove anything unused. Consider whether per-view decorators could replace middleware that only applies to specific URL patterns.
The layered architecture of middleware also makes debugging harder — an unexpected header might come from any layer. Logging the middleware class name alongside any modifications helps trace issues.
The one thing to remember: Django middleware is a chain of nested handlers that wraps every request/response — understanding the construction order, async boundaries, and exception flow is essential for writing middleware that’s both correct and performant.
See Also
- Python Django Admin Get an intuitive feel for Django Admin so Python behavior stops feeling unpredictable.
- Python Django Basics Get an intuitive feel for Django Basics so Python behavior stops feeling unpredictable.
- Python Django Celery Integration Why your Django app needs a helper to handle slow jobs in the background.
- Python Django Channels Websockets How Django can send real-time updates to your browser without you refreshing the page.
- Python Django Custom Management Commands How to teach Django new tricks by creating your own command-line shortcuts.