Management commands personalizados en Django

Estructura de un management command

Los management commands se ubican en el directorio management/commands/ de la aplicación:

catalogo/
    management/
        __init__.py
        commands/
            __init__.py
            importar_productos.py
            limpiar_sesiones_expiradas.py
            generar_informe.py

BaseCommand básico

# catalogo/management/commands/saludar.py
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = 'Muestra un saludo personalizable'

    def add_arguments(self, parser):
        parser.add_argument('nombre', type=str, help='Nombre a saludar')
        parser.add_argument(
            '--veces', '-n',
            type=int,
            default=1,
            help='Número de veces que se repite el saludo (por defecto: 1)'
        )
        parser.add_argument(
            '--mayusculas',
            action='store_true',
            help='Mostrar el saludo en mayúsculas'
        )

    def handle(self, *args, **options):
        nombre = options['nombre']
        veces = options['veces']
        mayusculas = options['mayusculas']

        for i in range(veces):
            saludo = f'¡Hola, {nombre}!'
            if mayusculas:
                saludo = saludo.upper()
            self.stdout.write(self.style.SUCCESS(saludo))

python manage.py saludar Ana --veces 3 --mayusculas

Command de importación de datos

# catalogo/management/commands/importar_productos.py
import csv
from pathlib import Path
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from catalogo.models import Producto, Categoria

class Command(BaseCommand):
    help = 'Importa productos desde un archivo CSV'

    def add_arguments(self, parser):
        parser.add_argument('archivo', type=str, help='Ruta al archivo CSV')
        parser.add_argument(
            '--dry-run',
            action='store_true',
            help='Simular la importación sin guardar datos'
        )
        parser.add_argument(
            '--actualizar',
            action='store_true',
            help='Actualizar productos existentes (por ISBN)'
        )

    def handle(self, *args, **options):
        archivo = Path(options['archivo'])
        dry_run = options['dry_run']
        actualizar = options['actualizar']

        if not archivo.exists():
            raise CommandError(f'El archivo {archivo} no existe.')

        creados = 0
        actualizados = 0
        errores = 0

        try:
            with open(archivo, 'r', encoding='utf-8') as f, transaction.atomic():
                reader = csv.DictReader(f)
                for numero_linea, fila in enumerate(reader, start=2):
                    try:
                        categoria, _ = Categoria.objects.get_or_create(
                            nombre=fila.get('categoria', 'Sin categoría')
                        )
                        datos = {
                            'nombre': fila['nombre'],
                            'precio': float(fila['precio']),
                            'categoria': categoria,
                            'activo': fila.get('activo', 'true').lower() == 'true',
                        }

                        if not dry_run:
                            if actualizar:
                                producto, fue_creado = Producto.objects.update_or_create(
                                    isbn=fila['isbn'],
                                    defaults=datos
                                )
                                if fue_creado:
                                    creados += 1
                                else:
                                    actualizados += 1
                            else:
                                Producto.objects.create(isbn=fila['isbn'], **datos)
                                creados += 1
                        else:
                            self.stdout.write(f'  [DRY RUN] {datos["nombre"]}')
                            creados += 1

                    except (KeyError, ValueError) as e:
                        self.stderr.write(
                            self.style.ERROR(f'Error en línea {numero_linea}: {e}')
                        )
                        errores += 1

                if dry_run:
                    raise Exception('DRY RUN: revirtiendo transacción')

        except Exception as e:
            if 'DRY RUN' not in str(e):
                raise CommandError(f'Error en la importación: {e}')

        if dry_run:
            self.stdout.write(self.style.WARNING(
                f'[DRY RUN] Se crearían {creados} productos. Usa sin --dry-run para aplicar.'
            ))
        else:
            self.stdout.write(self.style.SUCCESS(
                f'Importación completada: {creados} creados, {actualizados} actualizados, {errores} errores.'
            ))

Command de limpieza y mantenimiento

# catalogo/management/commands/limpiar_datos.py
from django.core.management.base import BaseCommand
from django.utils import timezone
from datetime import timedelta

class Command(BaseCommand):
    help = 'Limpia datos obsoletos de la base de datos'

    def add_arguments(self, parser):
        parser.add_argument('--dias', type=int, default=30, help='Días de antigüedad')
        parser.add_argument('--modelos', nargs='+', choices=['sesiones', 'logs', 'notificaciones'])

    def handle(self, *args, **options):
        dias = options['dias']
        modelos = options.get('modelos') or ['sesiones', 'logs', 'notificaciones']
        fecha_limite = timezone.now() - timedelta(days=dias)

        if 'sesiones' in modelos:
            from django.contrib.sessions.backends.db import SessionStore
            from django.contrib.sessions.models import Session
            eliminadas = Session.objects.filter(expire_date__lt=timezone.now()).delete()[0]
            self.stdout.write(f'Sesiones eliminadas: {eliminadas}')

        if 'notificaciones' in modelos:
            from usuarios.models import Notificacion
            eliminadas = Notificacion.objects.filter(
                leida=True, fecha__lt=fecha_limite
            ).delete()[0]
            self.stdout.write(f'Notificaciones eliminadas: {eliminadas}')

        self.stdout.write(self.style.SUCCESS('Limpieza completada.'))

Estilos de salida

self.stdout.write(self.style.SUCCESS('Operación completada ✓'))
self.stdout.write(self.style.WARNING('Advertencia: datos incompletos'))
self.stdout.write(self.style.ERROR('Error: operación fallida'))
self.stdout.write(self.style.NOTICE('Información adicional'))
self.stderr.write('Error en stderr')

Ejecutar desde código Python

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

# Desde una vista o señal
call_command('importar_productos', 'datos.csv', '--actualizar')

# Capturar la salida
salida = StringIO()
call_command('limpiar_datos', '--dias', '7', stdout=salida)
resultado = salida.getvalue()

Los management commands son la herramienta ideal para tareas de mantenimiento que se ejecutan periódicamente con cron, scripts de importación de datos iniciales, y cualquier operación administrativa que no necesita una interfaz web.