Django Advanced #2: Custom management commands

Framework Django management commands
8 min read

In #1 Async views you saw the async camp; this time we look at another axis of operations — command-line work. The home for batch jobs, data migrations, and maintenance scripts that you trigger with one word: manage.py.

What manage.py actually is #

The manage.py that ships automatically when you create a Django project is:

manage.py 
import os, sys
from django.core.management import execute_from_command_line

if __name__ == "__main__":
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
    execute_from_command_line(sys.argv)

The core is execute_from_command_line. It looks at your arguments, finds the matching command, and runs it. runserver, migrate, shell — all the same mechanism.

See built-in commands #

All commands 
python manage.py help

Partial output:

[django]
    check
    compilemessages
    createcachetable
    dbshell
    diffsettings
    dumpdata
    flush
    inspectdb
    loaddata
    makemessages
    makemigrations
    migrate
    runserver
    sendtestemail
    shell
    showmigrations
    sqlflush
    sqlmigrate
    sqlsequencereset
    squashmigrations
    startapp
    startproject
    test
    testserver

Beyond [django], you see groups per installed app ([myapp]). Under myapp is exactly where commands you create belong.

Your first custom command #

A command file must live in a fixed location to be discovered automatically.

Directory layout 
myapp/
├── __init__.py
├── models.py
├── views.py
└── management/
    ├── __init__.py
    └── commands/
        ├── __init__.py
        └── cleanup.py        # ← this file is the command

Both management/__init__.py and management/commands/__init__.py may be empty, but must exist. Without them, the command won’t be discovered.

myapp/management/commands/cleanup.py 
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = "Clean up expired tokens."

    def handle(self, *args, **options):
        self.stdout.write("Starting cleanup...")
        # actual work
        self.stdout.write(self.style.SUCCESS("Done"))

That’s all.

Run 
python manage.py cleanup

Rules:

  • The filename is the command name (cleanup.py → cleanup)
  • The file must contain a class named Command
  • It subclasses BaseCommand
  • handle(self, *args, **options) is the body
  • help is the description shown by manage.py help cleanup

Argument handling — add_arguments #

It runs on top of argparse.

Various argument kinds 
class Command(BaseCommand):
    help = "Compute per-user stats."

    def add_arguments(self, parser):
        # positional — required
        parser.add_argument("user_id", type=int, help="Target user ID")

        # optional — takes a value
        parser.add_argument(
            "--since",
            type=str,
            default="2026-01-01",
            help="Start date (YYYY-MM-DD)",
        )

        # flag — on/off switch
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="Simulate only, no actual changes",
        )

        # multiple values
        parser.add_argument(
            "--tags",
            nargs="+",
            help="Target tags (space-separated)",
        )

    def handle(self, *args, **options):
        user_id = options["user_id"]
        since = options["since"]
        dry_run = options["dry_run"]
        tags = options["tags"] or []
        self.stdout.write(f"user={user_id} since={since} dry={dry_run} tags={tags}")
Call example 
python manage.py compute_stats 42 --since 2026-04-01 --dry-run --tags python django

- in option names is automatically converted to _ in the options dict keys (--dry-run → options["dry_run"]).

Output — self.stdout, self.style #

Why use self.stdout.write instead of print():

  • You can swap stdout in tests (covered below)
  • The --no-color option is handled automatically
  • Windows console compatibility

Colors come from self.style:

Styles 
self.stdout.write(self.style.SUCCESS("OK"))           # green
self.stdout.write(self.style.WARNING("Caution"))      # yellow
self.stdout.write(self.style.ERROR("Failed"))         # red
self.stdout.write(self.style.NOTICE("Notice"))        # cyan
self.stdout.write(self.style.HTTP_INFO("INFO"))       # cyan
self.stderr.write(self.style.ERROR("Errors go to stderr"))

By convention, errors go to self.stderr. Shell redirection like 2> err.log then works naturally.

Exit — CommandError #

If something goes wrong and you need to abort, raise CommandError.

Error exit 
from django.core.management.base import BaseCommand, CommandError
from myapp.models import User

class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("user_id", type=int)

    def handle(self, *args, **options):
        try:
            user = User.objects.get(pk=options["user_id"])
        except User.DoesNotExist:
            raise CommandError(f"User {options['user_id']} not found.")

        self.stdout.write(self.style.SUCCESS(f"Found: {user.email}"))

CommandError prints in red to stderr and exits with code 1. cron failure notifications and systemd’s OnFailure= can pick it up.

CommandError is more idiomatic in Django than sys.exit(1).

Real-world — expired token cleanup #

The kind you’ll write most often in production.

myapp/management/commands/cleanup_tokens.py 
from datetime import timedelta
from django.core.management.base import BaseCommand
from django.utils import timezone
from myapp.models import AuthToken

class Command(BaseCommand):
    help = "Delete expired auth tokens."

    def add_arguments(self, parser):
        parser.add_argument(
            "--days",
            type=int,
            default=30,
            help="Treat tokens older than this many days as expired (default 30)",
        )
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="Count only, don't delete",
        )

    def handle(self, *args, **options):
        days = options["days"]
        cutoff = timezone.now() - timedelta(days=days)

        qs = AuthToken.objects.filter(created_at__lt=cutoff)
        count = qs.count()

        if options["dry_run"]:
            self.stdout.write(
                self.style.NOTICE(f"[dry-run] {count} candidates for deletion")
            )
            return

        deleted, _ = qs.delete()
        self.stdout.write(
            self.style.SUCCESS(f"Deleted {deleted} tokens (cutoff={cutoff:%Y-%m-%d})")
        )

The --dry-run pattern is strongly recommended for almost any cleanup command. It’s safer to simulate once before touching production data.

Real-world — batch stats computation #

myapp/management/commands/compute_daily_stats.py 
from django.core.management.base import BaseCommand
from django.db.models import Count, Sum
from django.utils import timezone
from myapp.models import Order, DailyStat

class Command(BaseCommand):
    help = "Compute daily stats up to yesterday and load into DailyStat."

    def add_arguments(self, parser):
        parser.add_argument("--date", type=str, help="YYYY-MM-DD")

    def handle(self, *args, **options):
        from datetime import datetime, date, timedelta

        target = (
            datetime.strptime(options["date"], "%Y-%m-%d").date()
            if options["date"]
            else date.today() - timedelta(days=1)
        )

        agg = Order.objects.filter(created_at__date=target).aggregate(
            count=Count("id"),
            total=Sum("amount"),
        )

        DailyStat.objects.update_or_create(
            date=target,
            defaults={
                "order_count": agg["count"] or 0,
                "total_amount": agg["total"] or 0,
            },
        )

        self.stdout.write(
            self.style.SUCCESS(
                f"{target}: {agg['count']} orders / {agg['total']}"
            )
        )

update_or_create — update if exists, create if not — is the core tool for making a command idempotent (safe to re-run). Even if cron failed once and you re-run, no duplicates appear.

Real-world — data migration #

Sometimes you need to leave the schema alone and change only the data (e.g., trim spaces from usernames, merge categories). Two paths:

  1. Django migration’s RunPython — small data transforms that ride along with schema changes
  2. A custom command — one-off, large data, or transforms that need staged progress

Big jobs fit a command. Example:

myapp/management/commands/normalize_emails.py 
from django.core.management.base import BaseCommand
from django.db import transaction
from myapp.models import User

class Command(BaseCommand):
    help = "Normalize emails to lowercase."

    def add_arguments(self, parser):
        parser.add_argument("--batch-size", type=int, default=1000)
        parser.add_argument("--dry-run", action="store_true")

    def handle(self, *args, **options):
        batch_size = options["batch_size"]
        dry_run = options["dry_run"]
        total = updated = 0

        qs = User.objects.exclude(email__exact="").iterator(chunk_size=batch_size)

        with transaction.atomic():
            for user in qs:
                total += 1
                normalized = user.email.lower().strip()
                if user.email == normalized:
                    continue
                user.email = normalized
                if not dry_run:
                    user.save(update_fields=["email"])
                updated += 1
                if total % batch_size == 0:
                    self.stdout.write(f"  progress {total}...")

        self.stdout.write(
            self.style.SUCCESS(
                f"checked {total} / changed {updated} ({'dry' if dry_run else 'commit'})"
            )
        )

iterator(chunk_size=...) streams rows instead of loading every row into memory. Essential for big tables. Covered more in #3.

call_command — invoking other commands #

When you want to call a command from code, or from another command.

from views.py or anywhere 
from django.core.management import call_command

call_command("cleanup_tokens", days=7, dry_run=True)
call_command("loaddata", "fixtures/seed.json")

Pass options as kwargs. --dry-run becomes dry_run=True.

Calling from inside another command #

Composite command 
class Command(BaseCommand):
    help = "Nightly batch — token cleanup + stats + reindex"

    def handle(self, *args, **options):
        call_command("cleanup_tokens", days=30)
        call_command("compute_daily_stats")
        self.stdout.write(self.style.SUCCESS("Nightly batch complete"))

Building several small commands and bundling them with a composite command works well in operations. Each can run independently; nightly, you run them all at once.

Pairing with cron / systemd #

cron #

/etc/cron.d/myproject 
# Daily at 3 AM
0 3 * * * www-data cd /opt/myproject && /opt/venv/bin/python manage.py cleanup_tokens >> /var/log/myproject/cleanup.log 2>&1

Rules:

  • Absolute paths — cron’s PATH is sparse. Spell out the python path
  • cd to the project directory — relative imports won’t break
  • Log both stdout and stderr — >> ... 2>&1
  • If env vars are needed, use BASH_ENV=/etc/profile, or export directly at the top of the cron file

systemd timer #

On modern Linux, a systemd timer is more visible and robust than cron.

/etc/systemd/system/cleanup.service 
[Unit]
Description=Django cleanup tokens

[Service]
Type=oneshot
User=www-data
WorkingDirectory=/opt/myproject
EnvironmentFile=/opt/myproject/.env
ExecStart=/opt/venv/bin/python manage.py cleanup_tokens
/etc/systemd/system/cleanup.timer 
[Unit]
Description=Daily cleanup

[Timer]
OnCalendar=daily
Persistent=true

[Install]
WantedBy=timers.target
Activate 
systemctl enable --now cleanup.timer
systemctl list-timers

Persistent=true catches up missed runs immediately after a server wakes back up. cron can’t do this.

Testing — capture output with StringIO #

call_command lets you specify stdout/stderr, which makes tests clean.

tests/test_cleanup.py 
from io import StringIO
from datetime import timedelta
from django.test import TestCase
from django.core.management import call_command
from django.utils import timezone
from myapp.models import AuthToken

class CleanupTokensTest(TestCase):
    def setUp(self):
        old = timezone.now() - timedelta(days=60)
        new = timezone.now()
        AuthToken.objects.create(value="old", created_at=old)
        AuthToken.objects.create(value="new", created_at=new)

    def test_dry_run_does_not_delete(self):
        out = StringIO()
        call_command("cleanup_tokens", dry_run=True, stdout=out)
        self.assertIn("1 candidates for deletion", out.getvalue())
        self.assertEqual(AuthToken.objects.count(), 2)

    def test_actual_delete(self):
        out = StringIO()
        call_command("cleanup_tokens", days=30, stdout=out)
        self.assertIn("Deleted 1 tokens", out.getvalue())
        self.assertEqual(AuthToken.objects.count(), 1)

It builds cleanly on top of the TestCase from Intermediate #7. The reason StringIO works with self.stdout.write but not with print is simple: stdout is swappable.

Common pitfalls #

1) Not discovered #

The most common cause is missing management/__init__.py or management/commands/__init__.py. They must exist, even if empty.

Also, the app must be in INSTALLED_APPS to be discovered.

2) BaseCommand.requires_migrations_checks #

Defaults to False. Setting True makes the command check for unapplied migrations before running and warn. Recommended for commands that touch production data.

Option 
class Command(BaseCommand):
    requires_migrations_checks = True

3) No automatic transaction #

Unlike views, commands have no automatic transaction like ATOMIC_REQUESTS. Commands that change data should wrap their work in with transaction.atomic(): themselves to be safe.

4) Signal explosion #

When touching large amounts of data, post_save signals can fire an external call per row. For migration-style work, temporarily disconnect signals or use bulk_update (see #5).

Wrap-up #

What you took home this time:

  • manage.py is the entry point of execute_from_command_line
  • Location: myapp/management/commands/<name>.py, both __init__.py files required
  • BaseCommand + handle(*args, **options), add_arguments(parser)
  • Arguments: positional, optional, flag (action="store_true"), nargs="+"
  • Output: self.stdout.write + self.style.SUCCESS/ERROR/...
  • Exit: CommandError (code 1, stderr)
  • Common kinds: expired cleanup, stat loading (update_or_create), data migration (iterator(chunk_size=...))
  • Call from code or other commands with call_command(...)
  • Pair with cron / systemd timer — absolute paths, log redirection
  • Testing: StringIO + call_command(..., stdout=out)
  • Pitfalls: missing __init__.py, no auto transaction, signal explosion

In the next post (#3 Query optimization) we layer on top of Intermediate #2’s ORM the full toolbox — N+1 diagnosis and fixes, select_related/prefetch_related, indexes, bulk_* — for the performance bottlenecks you hit most often in production.

X