Django Signals and Custom Middleware: Production Patterns

Signals and middleware are the "magic" of Django — and also where many production bugs hide. Here's how to use them well and when to reach for alternatives.

Signals: The Double-Edged Sword

Signals look elegant — "when X happens, do Y somewhere else." But they: - Hide control flow (grep won't find the connection) - Make debugging harder - Can cause subtle ordering bugs - Run synchronously in the request/response cycle

When signals are the right tool

• Cross-app hooks: Different apps reacting to the same event (e.g., user creation triggers profile creation in another app)
• Third-party integrations: When you don't control the sender
• Async messaging: Via post_save → Celery task

When they're the wrong tool

• Simple logic tied to a specific model → just override save()
• Things within a single app → call the function directly
• Heavy computation → signals run inline and slow requests

Practical Signal Patterns

Post-save with transaction safety

from django.db.models.signals import post_save
from django.db import transaction
from django.dispatch import receiver

@receiver(post_save, sender=Order)
def notify_on_order_created(sender, instance, created, **kwargs):
    if not created:
        return

    # Wait until the transaction commits before sending
    transaction.on_commit(lambda: send_order_email.delay(instance.id))

Always use transaction.on_commit() for side effects that depend on the object existing — otherwise your Celery task might run before the DB transaction commits.

Creating related objects

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(user=instance)

Better alternative — use get_or_create on access:

class User(AbstractUser):
    @property
    def profile(self):
        profile, _ = Profile.objects.get_or_create(user=self)
        return profile

Or better yet, create the profile explicitly where you create the user.

Pre-save for computed fields

@receiver(pre_save, sender=Article)
def set_slug(sender, instance, **kwargs):
    if not instance.slug:
        instance.slug = slugify(instance.title)

Or just override save():

class Article(models.Model):
    def save(self, *args, **kwargs):
        if not self.slug:
            self.slug = slugify(self.title)
        super().save(*args, **kwargs)

I prefer override — it's explicit, testable, and findable.

Signal connection

# apps.py
class MyappConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        from . import signals  # noqa

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender='myapp.Order')  # string to avoid circular imports
def my_handler(sender, instance, **kwargs):
    ...

Custom signals

# signals.py
from django.dispatch import Signal

order_completed = Signal()  # providing_args is deprecated

# In views/services
order_completed.send(sender=Order, instance=order, amount=order.total)

# Receivers
@receiver(order_completed)
def on_complete(sender, instance, amount, **kwargs):
    ...

Signal Debugging

Signals running silently in production are a nightmare. Add logging:

import logging
logger = logging.getLogger(__name__)

@receiver(post_save, sender=Order)
def log_order_save(sender, instance, created, **kwargs):
    logger.info(
        "Order saved",
        extra={'order_id': instance.id, 'created': created}
    )

List all connected signals:

from django.db.models.signals import post_save

for receiver_func, _ in post_save.receivers:
    print(receiver_func)

Middleware: The Request Lifecycle

Every request passes through middleware before the view and on the way back:

Request → MW1.__call__ → MW2.__call__ → View → MW2 (after) → MW1 (after) → Response

Writing middleware

Modern function-based middleware:

def simple_middleware(get_response):
    # One-time configuration at startup
    def middleware(request):
        # Code before the view
        response = get_response(request)
        # Code after the view
        return response
    return middleware

Class-based for more control:

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

    def __call__(self, request):
        start = time.time()
        response = self.get_response(request)
        duration = time.time() - start
        response['X-Response-Time'] = f'{duration:.3f}s'
        return response

    def process_view(self, request, view_func, view_args, view_kwargs):
        # Called before the view
        pass

    def process_exception(self, request, exception):
        # Called if view raises
        pass

    def process_template_response(self, request, response):
        # Called if response has render()
        return response

Production Middleware Patterns

Request IDs for tracing

import uuid
import logging

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

    def __call__(self, request):
        request.id = request.headers.get('X-Request-ID') or str(uuid.uuid4())

        # Push to logging context
        logging.LoggerAdapter(
            logging.getLogger(),
            {'request_id': request.id}
        )

        response = self.get_response(request)
        response['X-Request-ID'] = request.id
        return response

Structured access logging

class AccessLogMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.logger = logging.getLogger('access')

    def __call__(self, request):
        start = time.perf_counter()
        response = self.get_response(request)
        duration_ms = (time.perf_counter() - start) * 1000

        self.logger.info(
            'request',
            extra={
                'method': request.method,
                'path': request.path,
                'status': response.status_code,
                'duration_ms': round(duration_ms, 2),
                'user_id': request.user.id if request.user.is_authenticated else None,
                'ip': self.get_client_ip(request),
            }
        )
        return response

    def get_client_ip(self, request):
        x_forwarded = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded:
            return x_forwarded.split(',')[0].strip()
        return request.META.get('REMOTE_ADDR')

Audit logging for sensitive actions

class AuditMiddleware:
    SENSITIVE_PATHS = ['/admin/', '/api/users/', '/api/payments/']

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

    def __call__(self, request):
        is_sensitive = any(request.path.startswith(p) for p in self.SENSITIVE_PATHS)
        response = self.get_response(request)

        if is_sensitive and request.method in ('POST', 'PUT', 'PATCH', 'DELETE'):
            AuditLog.objects.create(
                user=request.user if request.user.is_authenticated else None,
                path=request.path,
                method=request.method,
                status=response.status_code,
                ip=request.META.get('REMOTE_ADDR'),
            )

        return response

Per-request database connection limits

For multi-tenant apps, prevent one request from holding a connection too long:

from django.db import connection

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

    def __call__(self, request):
        response = self.get_response(request)
        connection.close_if_unusable_or_obsolete()
        return response

Feature flags middleware

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

    def __call__(self, request):
        request.features = self.load_flags(request.user)
        return self.get_response(request)

    def load_flags(self, user):
        if not user.is_authenticated:
            return set()
        cached = cache.get(f'flags:{user.id}')
        if cached is None:
            cached = set(user.feature_flags.values_list('name', flat=True))
            cache.set(f'flags:{user.id}', cached, 300)
        return cached

Middleware Ordering Matters

The order in MIDDLEWARE is execution order. Common mistakes:

# Wrong: Session middleware after auth middleware
MIDDLEWARE = [
    'django.contrib.auth.middleware.AuthenticationMiddleware',  # needs session!
    'django.contrib.sessions.middleware.SessionMiddleware',
]

# Correct order (simplified)
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',
    # Your custom middleware here (mostly)
]

Async Middleware (Django 4.1+)

import asyncio
from asgiref.sync import iscoroutinefunction

class AsyncCapableMiddleware:
    async_capable = True
    sync_capable = True

    def __init__(self, get_response):
        self.get_response = get_response
        self.async_mode = iscoroutinefunction(get_response)

    def __call__(self, request):
        if self.async_mode:
            return self.__acall__(request)
        response = self.get_response(request)
        return response

    async def __acall__(self, request):
        response = await self.get_response(request)
        return response

Testing Middleware

from django.test import RequestFactory, TestCase

class MiddlewareTestCase(TestCase):
    def setUp(self):
        self.factory = RequestFactory()

    def test_timing_middleware(self):
        def get_response(request):
            return HttpResponse('ok')

        middleware = TimingMiddleware(get_response)
        request = self.factory.get('/')
        response = middleware(request)
        self.assertIn('X-Response-Time', response)

Common Pitfalls

Thread safety

Don't store state on self:

# Bad: shared across requests in same worker
class BadMiddleware:
    def __init__(self, get_response):
        self.current_user = None  # DANGER

    def __call__(self, request):
        self.current_user = request.user  # race condition!

Blocking the event loop

Don't do heavy I/O in async middleware synchronously. Either use async alternatives or move work to a task queue.

Swallowing exceptions

# Bad: hides the real error
def __call__(self, request):
    try:
        return self.get_response(request)
    except Exception:
        return HttpResponse('Error')  # no logging, no traceback

Summary

• Signals: use sparingly, prefer explicit calls. Wrap side effects in transaction.on_commit.
• Middleware: great for cross-cutting concerns (logging, auth, timing, tracing).
• Order matters. Document it. Test it.
• Async? Plan for it from the start.
• Log everything in production.

Done right, signals and middleware make your codebase more extensible. Done wrong, they make debugging a nightmare. Pick them for the right reasons.