[Django] Management Commands

2026년 6월 20일 · 수정 2026년 6월 20일 · 📖 약 1분 · 356자/단어 #django #management #command #cli

정의

Django 커스텀 명령은 manage.py <command> 형태로 실행되는 CLI 도구. 정기 작업(cron + management command), 데이터 마이그레이션, import/export, 디버깅 등에 사용. argparse 기반.

위치

myapp/
    management/
        __init__.py
        commands/
            __init__.py
            cleanup.py
            import_users.py

management/commands/ 안의 각 .py 파일이 명령.

최소 예시

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

class Command(BaseCommand):
    help = "Says hello"

    def handle(self, *args, **options):
        self.stdout.write("Hello, world!")

python manage.py hello

인자

class Command(BaseCommand):
    help = "Import users from CSV"

    def add_arguments(self, parser):
        # 위치 인자
        parser.add_argument("filename", type=str, help="CSV file path")

        # 옵션
        parser.add_argument("--dry-run", action="store_true")
        parser.add_argument("--batch-size", type=int, default=1000)
        parser.add_argument(
            "--update-existing",
            action="store_true",
            help="Update existing users",
        )

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

        self.stdout.write(f"Importing from {filename}")
        if dry_run:
            self.stdout.write(self.style.WARNING("DRY RUN"))
        ...

python manage.py import_users data.csv --dry-run --batch-size 500

출력 (스타일)

self.stdout.write("normal")
self.stdout.write(self.style.SUCCESS("✓ Done"))
self.stdout.write(self.style.ERROR("✗ Failed"))
self.stdout.write(self.style.WARNING("warn"))
self.stdout.write(self.style.NOTICE("notice"))
self.stdout.write(self.style.SQL_FIELD("field"))
self.stderr.write(self.style.ERROR("error to stderr"))

색상 자동 (TTY일 때). 색상 끄려면 --no-color.

진행률 / 로그

import logging
logger = logging.getLogger(__name__)

def handle(self, *args, **options):
    items = Item.objects.all()
    total = items.count()
    for i, item in enumerate(items, 1):
        if i % 100 == 0:
            self.stdout.write(f"Processed {i}/{total}")
        self.process(item)
    self.stdout.write(self.style.SUCCESS(f"Done: {total} items"))

또는 tqdm:

from tqdm import tqdm

for item in tqdm(items, desc="Processing"):
    self.process(item)

예외 처리

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

def handle(self, *args, **options):
    if not Path(options["filename"]).exists():
        raise CommandError(f"File not found: {options['filename']}")
    try:
        self.process()
    except Exception as e:
        raise CommandError(f"Processing failed: {e}") from e

CommandError는 Django가 깔끔하게 stderr로.

verbosity

python manage.py mycmd -v 0    # 최소
python manage.py mycmd -v 1    # 기본
python manage.py mycmd -v 2    # 상세
python manage.py mycmd -v 3    # 매우 상세

def handle(self, *args, **options):
    verbosity = options["verbosity"]
    if verbosity >= 2:
        self.stdout.write("Detailed info")

트랜잭션

from django.db import transaction

class Command(BaseCommand):
    @transaction.atomic
    def handle(self, *args, **options):
        ...    # 전체가 한 트랜잭션, 예외 시 rollback

또는 명시적:

def handle(self, *args, **options):
    with transaction.atomic():
        for batch in batches:
            self.process(batch)

자주 보는 명령 패턴

데이터 import

import csv
from django.core.management.base import BaseCommand
from myapp.models import User

class Command(BaseCommand):
    help = "Import users from CSV"

    def add_arguments(self, parser):
        parser.add_argument("filename")
        parser.add_argument("--dry-run", action="store_true")

    @transaction.atomic
    def handle(self, *args, **options):
        with open(options["filename"]) as f:
            reader = csv.DictReader(f)
            count = 0
            for row in reader:
                user, created = User.objects.get_or_create(
                    email=row["email"],
                    defaults={"name": row["name"]},
                )
                if created:
                    count += 1
                    self.stdout.write(f"Created: {user.email}")

            if options["dry_run"]:
                transaction.set_rollback(True)
                self.stdout.write(self.style.WARNING(f"DRY RUN: would create {count}"))
            else:
                self.stdout.write(self.style.SUCCESS(f"Created {count} users"))

Cleanup

class Command(BaseCommand):
    help = "Cleanup old data"

    def add_arguments(self, parser):
        parser.add_argument("--days", type=int, default=30)

    def handle(self, *args, **options):
        cutoff = timezone.now() - timedelta(days=options["days"])
        deleted, _ = Post.objects.filter(
            is_draft=True, created_at__lt=cutoff
        ).delete()
        self.stdout.write(f"Deleted {deleted} drafts older than {options['days']} days")

Cron으로:

0 3 * * * cd /app && python manage.py cleanup --days 30

Send periodic emails

class Command(BaseCommand):
    help = "Send daily digest"

    def handle(self, *args, **options):
        for user in User.objects.filter(daily_digest=True):
            from myapp.tasks import send_digest
            send_digest.delay(user.id)
        self.stdout.write(self.style.SUCCESS("Queued all digest jobs"))

Re-index search

class Command(BaseCommand):
    help = "Reindex search"

    def handle(self, *args, **options):
        from myapp.search import reindex
        for post in Post.objects.all().iterator(chunk_size=1000):
            reindex(post)
        self.stdout.write(self.style.SUCCESS("Done"))

표준 명령 활용

Django 기본 제공 (django.core.management):

runserver, shell, dbshell
migrate, makemigrations, showmigrations, sqlmigrate, squashmigrations
startproject, startapp
createsuperuser, changepassword
collectstatic, findstatic
dumpdata, loaddata
test, check
clearsessions
compilemessages, makemessages

django-extensions로 추가:

shell_plus (auto import models)
runserver_plus (Werkzeug debugger)
graph_models (DB 다이어그램)
show_urls
print_settings

호출 방법

CLI 외에 Python 코드에서도:

from django.core.management import call_command

call_command("migrate")
call_command("import_users", "data.csv", dry_run=True)

테스트에서 명령 자체 검증.

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

class MyCommandTest(TestCase):
    def test_import(self):
        out = StringIO()
        call_command("import_users", "test.csv", stdout=out)
        self.assertIn("Created", out.getvalue())

함정

1. command 없음

python manage.py mycmd
# CommandError: Unknown command

management/__init__.py 있는지 (빈 파일)
management/commands/__init__.py 있는지
INSTALLED_APPS에 앱 등록됐는지
파일명이 _/- 등 명확한지

2. 큰 데이터셋 메모리

for item in Item.objects.all():    # 전부 로드
    process(item)

# 대신
for item in Item.objects.all().iterator(chunk_size=2000):
    process(item)

3. 트랜잭션 + signal

@transaction.atomic 내부에서 외부 API 호출은 트랜잭션 길어짐 → DB 락. 외부 호출은 트랜잭션 밖.

4. 환경 분리

def handle(self, *args, **options):
    if not settings.DEBUG and not options["force"]:
        raise CommandError("Production: use --force")

위험한 명령은 안전장치.

5. cron + 환경변수

0 3 * * * /usr/bin/python /app/manage.py cleanup

cron 환경은 PATH, ENV가 제한적. shell wrapper 또는 절대 경로 + dotenv 명시 로드.