Middleware, signals and management commands

Run code around every request, react to model changes without hiding logic, and package operations as repeatable manage.py commands.

Writing middleware

Middleware is a chain of callables wrapped around the view. The request passes down the list in order; the response passes back up in reverse. That single rule explains almost every ordering bug: authentication must run before anything that reads request.user, and anything that touches the response headers must sit high enough to see responses from the layers above it.

# blog/middleware.py
import time

class RequestTimingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response        # called once, at startup

    def __call__(self, request):
        started = time.perf_counter()
        response = self.get_response(request)   # do not wrap in try/except and swallow
        response["X-Response-Time"] = f"{time.perf_counter() - started:.3f}s"
        return response

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

    def __call__(self, request):
        host = request.get_host().split(":")[0]
        request.tenant = resolve_tenant(host)   # attribute convention: no leading underscore
        if request.tenant is None:
            return HttpResponseNotFound("Unknown host")   # short-circuit: skip the view
        return self.get_response(request)

# config/settings.py
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",
    "blog.middleware.RequestTimingMiddleware",
]

• __init__ runs once per process; never do per-request work there, and never keep request state on self — the same instance serves concurrent requests.
• Returning a response from __call__ before calling get_response short-circuits the chain: later middleware never runs and the view is never reached. Useful for redirects, maintenance pages and tenant resolution.
• __call__ may also be declared async def to take part in the async chain; Django adapts sync and async middleware at the boundary, but a blocking database call inside async middleware still blocks the event loop.
• Order matters and is visible in settings.MIDDLEWARE. Add your own after the security and session layers, and put CsrfViewMiddleware before anything that reads the POST body for authentication.

Signals, and when they hurt

# blog/signals.py
from django.db.models.signals import post_delete, post_save
from django.dispatch import receiver
from .models import Post

@receiver(post_save, sender=Post, dispatch_uid="blog.notify_published")
def notify_published(sender, instance, created, **kwargs):
    if not created and instance.status == Post.Status.PUBLISHED:
        enqueue_newsletter(instance.pk)

@receiver(post_delete, sender=Post, dispatch_uid="blog.cleanup_files")
def cleanup_files(sender, instance, **kwargs):
    if instance.cover:
        instance.cover.delete(save=False)

# blog/apps.py — import the module so the receivers are registered
from django.apps import AppConfig

class BlogConfig(AppConfig):
    name = "blog"

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

• Signals fire per instance through the ORM. queryset.update(), bulk_create(), bulk_update() and raw SQL bypass every model signal, so a workflow that depends on one is silently skipped by any bulk operation.
• dispatch_uid makes registration idempotent, which matters because a module imported twice would otherwise connect the receiver twice and run the handler twice.
• A signal that saves the same model inside its own post_save handler recurses. Guard with update_fields, a flag on the instance, or keep the write out of the handler.

Management commands

# blog/management/commands/expire_drafts.py
from django.core.management.base import BaseCommand, CommandError
from django.utils import timezone
from datetime import timedelta
from blog.models import Post

class Command(BaseCommand):
    help = "Delete or archive drafts older than N days."

    def add_arguments(self, parser):
        parser.add_argument("--days", type=int, default=90)
        parser.add_argument("--archive", action="store_true",
                            help="Set status instead of deleting")

    def handle(self, *args, **options):
        cutoff = timezone.now() - timedelta(days=options["days"])
        qs = Post.objects.filter(status=Post.Status.DRAFT, created_at__lt=cutoff)
        count = qs.count()
        if count == 0:
            self.stdout.write("Nothing to do.")
            return
        qs.update(status=Post.Status.ARCHIVED) if options["archive"] else qs.delete()
        self.stdout.write(self.style.SUCCESS(f"{count} drafts processed"))

python manage.py expire_drafts --days 30 --archive
# cron: 17 4 * * * /srv/app/.venv/bin/python /srv/app/manage.py expire_drafts --days 30

• The command class must sit in <app>/management/commands/<name>.py with __init__.py files at each level; the file name becomes the command name.
• Use self.stdout.write rather than print, so output respects --verbosity and can be redirected; raise CommandError for a user-facing failure, which exits non-zero and prints a clean message.
• In tests call call_command("expire_drafts", days=1) and assert the resulting rows. The command is ordinary Python, so it needs no shell.

FAQ

Related

Settings, caching and query performance Deployment: gunicorn, static files and the security checklist

Last refreshed 2026-09-18.