Django Custom Management Commands

TL;DR

Django allows custom management commands to extend the core manage.py interface, allowing easy integration with the application when building backend processes, automation scripts, and scheduled jobs (while providing access into the Django application environment, data model and functions via the same structures used to build the website).

Where to create and/or find the code

Django discovers management commands through a specific directory layout in your apps:

your_app/
    management/
        __init__.py
        commands/
            __init__.py
            send_notifications.py
            update_reports.py

Every command file defines a Command class that extends BaseCommand:

from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = 'Send notifications to users for new reports'

    def handle(self, *args, **options):
        # Command logic goes here
        self.stdout.write('Task completed')

Run the command with python manage.py send_notifications. The filename becomes the command name.

Adding Command-Line Arguments

The add_arguments() method configures your command’s interface using Python’s argparse:

def add_arguments(self, parser):
    parser.add_argument(
        '--dry-run',
        action='store_true',
        help='Run without making changes',
    )

    parser.add_argument(
        '--force',
        action='store_true',
        help='Run all active searches regardless of schedule',
    )

Access arguments through the options dictionary in handle():

def handle(self, *args, **options):
    dry_run = options['dry_run']
    force = options['force']

    if dry_run:
        self.stdout.write(self.style.WARNING('Running in DRY-RUN mode'))

Proper Output and Styling

Always use self.stdout and self.stderr instead of print statements. This ensures output is captured correctly during testing and allows output redirection:

# Good practices
self.stdout.write(f'Processing user: {user.username}')
self.stdout.write(self.style.SUCCESS('Task completed successfully'))
self.stdout.write(self.style.WARNING('Running in dry-run mode'))
self.stderr.write(self.style.ERROR('Failed to process item'))

The style attribute provides colored output that works across terminals: SUCCESS, WARNING, ERROR, NOTICE, and SQL_KEYWORD.

Example: Scheduled Notifications

Here’s a simplified version of a notification command that sends emails about new reports:

from django.core.management.base import BaseCommand
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
from django.conf import settings

from tracker.models import Report, NotificationLog

User = get_user_model()

class Command(BaseCommand):
    help = 'Send notifications to users for new reports'

    def add_arguments(self, parser):
        parser.add_argument(
            '--dry-run',
            action='store_true',
            help='Show what would be sent without sending',
        )

    def handle(self, *args, **options):
        dry_run = options['dry_run']
        total_sent = 0

        for user in User.objects.filter(is_active=True):
            # Find new reports since last notification
            last_notification = NotificationLog.objects.filter(
                user=user
            ).order_by('-sent_at').first()

            if last_notification:
                new_reports = Report.objects.filter(
                    user=user,
                    created_at__gt=last_notification.sent_at
                )
            else:
                new_reports = Report.objects.filter(user=user)[:10]

            if not new_reports.exists():
                continue

            self.stdout.write(f'Found {new_reports.count()} reports for {user.username}')

            if dry_run:
                self.stdout.write(self.style.WARNING(
                    f'  [DRY-RUN] Would send email to {user.email}'
                ))
            else:
                send_mail(
                    subject='New Reports Available',
                    message=f'You have {new_reports.count()} new reports.',
                    from_email=settings.DEFAULT_FROM_EMAIL,
                    recipient_list=[user.email],
                )
                total_sent += 1
                self.stdout.write(self.style.SUCCESS(
                    f'  Sent notification to {user.email}'
                ))

        self.stdout.write(self.style.SUCCESS(
            f'Completed: {total_sent} notifications sent'
        ))

Schedule this with cron:

# Run every hour
0 * * * * cd /path/to/project && python manage.py send_notifications

Example: Async Report Generation

Commands can use async functionality for tasks like web scraping or API calls:

import anyio
from django.core.management.base import BaseCommand
from django.utils import timezone
from datetime import timedelta

from tracker.models import ProductSearch, Report

class Command(BaseCommand):
    help = 'Run due product searches and generate reports'

    def add_arguments(self, parser):
        parser.add_argument(
            '--force',
            action='store_true',
            help='Run all searches regardless of schedule',
        )

    async def fetch_product_data(self, search):
        """Async method to fetch product information."""
        # Use async HTTP client, browser automation, etc.
        pass

    def handle(self, **options):
        force = options.get('force', False)
        command_start = timezone.now()

        searches = ProductSearch.objects.filter(is_active=True)

        for search in searches:
            # Check if search is due
            if not force and search.last_searched:
                time_since = command_start - search.last_searched
                if time_since < timedelta(hours=search.frequency_hours):
                    continue

            self.stdout.write(f'Processing: {search.product_name}')

            try:
                # Run async task
                result = anyio.run(self.fetch_product_data, search)

                # Create report
                Report.objects.create(
                    user=search.user,
                    content=result,
                    report_type='price_check'
                )

                # Update last run time
                search.last_searched = command_start
                search.save()

                self.stdout.write(self.style.SUCCESS('  Report created'))

            except Exception as e:
                self.stderr.write(self.style.ERROR(f'  Failed: {e}'))

Error Handling

Raise CommandError for expected problems that should terminate the command:

from django.core.management.base import BaseCommand, CommandError

class Command(BaseCommand):
    def handle(self, *args, **options):
        if not settings.EMAIL_CONFIGURED:
            raise CommandError('Email not configured. Set EMAIL_HOST in settings.')

This displays a clean error message without a full traceback. For unexpected errors, let them propagate naturally so you see the full context during development.

Commands and Transactions

By default, Django wraps each command execution in a database transaction (when using a transactional database). For long-running commands that should commit incrementally, use the @transaction.atomic decorator on specific sections:

from django.db import transaction

class Command(BaseCommand):
    def handle(self, *args, **options):
        for batch in get_batches():
            with transaction.atomic():
                # This batch commits independently
                process_batch(batch)

Testing Commands

Test commands using Django’s call_command() helper:

from io import StringIO
from django.core.management import call_command
from django.test import TestCase

class CommandTests(TestCase):
    def test_dry_run_mode(self):
        out = StringIO()
        call_command('send_notifications', '--dry-run', stdout=out)
        self.assertIn('DRY-RUN', out.getvalue())

        # Verify no emails were sent
        self.assertEqual(len(mail.outbox), 0)

When to Use Management Commands

Management commands excel for:

Scheduled jobs: Daily reports, cleanup tasks, data synchronization
Batch processing: Import/export operations, bulk updates
System integration: Connecting Django with external services or APIs
Development tools: Database seeding, cache warming, test data generation
Administrative tasks: User management, permission updates, maintenance operations

They provide Django’s full environment including ORM access, settings, and logging while staying outside the web request cycle.

Summary

Custom management commands extend Django’s manage.py interface to handle backend processes and automation. They provide argument parsing, proper error handling, styled output, and full access to your Django application’s models and configuration.

The commands in this post demonstrate patterns for notification systems and scheduled report generation, but the same approach applies to any task that needs Django’s ORM and settings without the request-response cycle.

Reference: Django Custom Management Commands

TL;DR#

Where to create and/or find the code#

Adding Command-Line Arguments#

Proper Output and Styling#

Example: Scheduled Notifications#

Example: Async Report Generation#

Error Handling#

Commands and Transactions#

Testing Commands#

When to Use Management Commands#

Summary#