장고 고급 #2 Custom management commands

프레임워크 장고 management commands
7 분 소요

#1 Async views에서 비동기 진영을 보았다면, 이번에는 운영의 또 한 축인 커맨드라인 작업입니다. manage.py 한 단어로 굴리는 배치 작업, 데이터 마이그레이션, 유지보수 스크립트를 다룹니다.

manage.py의 정체 #

장고 프로젝트를 만들면 자동으로 따라오는 manage.py는:

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)

핵심은 execute_from_command_line. 그게 인자를 보고 적절한 커맨드를 찾아 실행합니다. runserver, migrate, shell 모두 같은 메커니즘입니다.

빌트인 커맨드 보기 #

모든 커맨드 
python manage.py help

출력 일부:

[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

[django] 외에도 설치된 앱 ([myapp]) 별로 그룹이 나옵니다. myapp 그룹에 우리가 만든 커맨드가 들어갈 자리입니다.

첫 커스텀 커맨드 #

커맨드 파일은 정해진 위치에 있어야 자동 발견됩니다.

디렉터리 구조 
myapp/
├── __init__.py
├── models.py
├── views.py
└── management/
    ├── __init__.py
    └── commands/
        ├── __init__.py
        └── cleanup.py        # ← 이 파일이 커맨드

management/__init__.py, management/commands/__init__.py 둘 다 비어 있어도 반드시 있어야 합니다. 없으면 발견 안 됩니다.

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

class Command(BaseCommand):
    help = "만료된 토큰을 청소합니다."

    def handle(self, *args, **options):
        self.stdout.write("청소 시작...")
        # 실제 작업
        self.stdout.write(self.style.SUCCESS("완료"))

위 코드가 커스텀 커맨드의 최소 구조입니다.

실행 
python manage.py cleanup

규칙:

  • 파일 이름이 곧 커맨드 이름 (cleanup.py → cleanup)
  • 파일 안에 Command 클래스가 있어야 함
  • BaseCommand 상속
  • handle(self, *args, **options)가 본체
  • help가 manage.py help cleanup의 설명

인자 처리 — add_arguments #

argparse 위에서 동작합니다.

여러 종류의 인자 
class Command(BaseCommand):
    help = "사용자별 통계를 계산합니다."

    def add_arguments(self, parser):
        # positional — 필수
        parser.add_argument("user_id", type=int, help="대상 사용자 ID")

        # optional — 값을 받음
        parser.add_argument(
            "--since",
            type=str,
            default="2026-01-01",
            help="시작 날짜 (YYYY-MM-DD)",
        )

        # flag — 켜고 끄는 스위치
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="실제 변경 없이 시뮬레이션만",
        )

        # 여러 값
        parser.add_argument(
            "--tags",
            nargs="+",
            help="대상 태그들 (공백 구분)",
        )

    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}")
호출 예 
python manage.py compute_stats 42 --since 2026-04-01 --dry-run --tags python django

옵션 이름의 -는 자동으로 _로 변환돼서 options dict의 키가 됩니다 (--dry-run → options["dry_run"]).

출력 — self.stdout, self.style #

print() 대신 **self.stdout.write**를 쓰는 이유:

  • 테스트할 때 stdout을 갈아 끼울 수 있음 (아래에서)
  • --no-color 옵션 자동 처리
  • Windows 콘솔 호환

색상은 self.style로:

스타일 
self.stdout.write(self.style.SUCCESS("OK"))           # 녹색
self.stdout.write(self.style.WARNING("주의"))          # 노랑
self.stdout.write(self.style.ERROR("실패"))            # 빨강
self.stdout.write(self.style.NOTICE("알림"))           # 청록
self.stdout.write(self.style.HTTP_INFO("INFO"))       # 청록
self.stderr.write(self.style.ERROR("에러는 stderr로"))

에러는 **self.stderr**로 보내는 게 관례. 2> err.log 같은 셸 리다이렉트가 자연스럽게 동작합니다.

종료 — CommandError #

문제가 생겨 종료해야 한다면 CommandError를 raise.

에러 종료 
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"사용자 {options['user_id']} 가 없습니다.")

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

CommandError는 빨강으로 stderr에 출력되고 종료 코드 1로 끝납니다. cron의 실패 알림이나 systemd의 OnFailure=가 잡을 수 있습니다.

sys.exit(1)보다 CommandError가 장고 관례.

실전 — 만료 토큰 청소 #

운영에서 가장 자주 만드는 부류.

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 = "만료된 인증 토큰을 삭제합니다."

    def add_arguments(self, parser):
        parser.add_argument(
            "--days",
            type=int,
            default=30,
            help="이 일수 이전의 토큰을 만료로 본다 (기본 30)",
        )
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="삭제 없이 카운트만",
        )

    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} 개 삭제 대상")
            )
            return

        deleted, _ = qs.delete()
        self.stdout.write(
            self.style.SUCCESS(f"{deleted} 개 토큰 삭제 완료 (cutoff={cutoff:%Y-%m-%d})")
        )

--dry-run 패턴은 거의 모든 정리 커맨드에 권장합니다. 운영 데이터를 건드리는 일은 한 번 시뮬레이션을 거치는 게 안전합니다.

실전 — 통계 일괄 계산 #

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 = "어제까지의 일별 통계를 계산해 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']} 건 / {agg['total']} 원"
            )
        )

update_or_create가 있으면 갱신, 없으면 생성 — 재실행해도 안전한 (idempotent) 커맨드를 만드는 핵심 도구. cron이 한 번 실패해서 다시 돌려도 중복이 안 생깁니다.

실전 — 데이터 마이그레이션 #

스키마는 그대로 두고 데이터만 바꿔야 할 때가 많습니다 (예: 사용자 이름의 공백 정리, 카테고리 통합 등). 두 갈래:

  1. 장고 마이그레이션의 RunPython — 스키마 변경과 함께 가는 작은 데이터 변환
  2. 커스텀 커맨드 — 일회성, 큰 데이터, 단계별 진행이 필요한 변환

큰 작업은 커맨드가 어울립니다. 예시:

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 = "이메일을 소문자로 정규화합니다."

    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"  진행 {total}...")

        self.stdout.write(
            self.style.SUCCESS(
                f"검사 {total} / 변경 {updated} ({'dry' if dry_run else 'commit'})"
            )
        )

iterator(chunk_size=...)가 메모리에 모든 행을 올리지 않고 스트리밍합니다. 큰 테이블에 필수. #3에서 더 다룹니다.

call_command — 다른 커맨드 호출 #

커맨드를 코드에서, 또는 다른 커맨드에서 부르고 싶을 때.

from views.py 또는 다른 곳 
from django.core.management import call_command

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

옵션은 kwargs로 전달. --dry-run은 dry_run=True.

다른 커맨드 안에서 부르기 #

조합 커맨드 
class Command(BaseCommand):
    help = "야간 일괄 작업 — 토큰 청소 + 통계 + 인덱스 재작성"

    def handle(self, *args, **options):
        call_command("cleanup_tokens", days=30)
        call_command("compute_daily_stats")
        self.stdout.write(self.style.SUCCESS("야간 일괄 완료"))

여러 작은 커맨드를 만들고 조합 커맨드로 묶는 게 운영에 편합니다. 각각 따로도 돌릴 수 있고, 야간엔 한 번에 정리합니다.

cron / systemd와 결합 #

cron #

/etc/cron.d/myproject 
# 매일 새벽 3시
0 3 * * * www-data cd /opt/myproject && /opt/venv/bin/python manage.py cleanup_tokens >> /var/log/myproject/cleanup.log 2>&1

규칙:

  • 절대경로 — cron 환경은 PATH가 빈약. 파이썬 경로 명시
  • 프로젝트 디렉터리로 cd — 상대 경로 import 안 깨짐
  • stdout/stderr 둘 다 로깅 — >> ... 2>&1
  • 환경 변수가 필요하면 BASH_ENV=/etc/profile 또는 cron 파일 상단에 직접 export

systemd timer #

요즘 리눅스는 systemd timer가 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
활성화 
systemctl enable --now cleanup.timer
systemctl list-timers

Persistent=true가 서버가 꺼져 있어 놓친 실행을 깨어난 직후 보충해줍니다. cron에는 없는 장점.

테스트 — StringIO로 출력 캡처 #

call_command에 stdout/stderr를 명시할 수 있어 테스트가 깔끔합니다.

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 개 삭제 대상", 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("1 개 토큰 삭제 완료", out.getvalue())
        self.assertEqual(AuthToken.objects.count(), 1)

중급 #7의 TestCase 위에 그대로 얹힙니다. StringIO가 print가 아닌 self.stdout.write를 쓰는 이유 — stdout 자리를 갈아 끼울 수 있어서.

자주 만나는 함정 #

1) 발견 안 됨 #

management/__init__.py, management/commands/__init__.py 누락이 가장 흔합니다. 둘 다 빈 파일이라도 있어야 합니다.

또 앱이 INSTALLED_APPS에 등록돼 있어야 발견됩니다.

2) BaseCommand.requires_migrations_checks #

기본값은 False. True로 두면 커맨드 실행 전 미적용 마이그레이션이 있는지 검사하고 경고합니다. 운영 데이터를 건드리는 커맨드는 켜두는 게 안전.

옵션 
class Command(BaseCommand):
    requires_migrations_checks = True

3) 트랜잭션 자동 안 걸림 #

view와 달리 커맨드는 ATOMIC_REQUESTS 같은 자동 트랜잭션이 없습니다. 데이터를 바꾸는 커맨드는 직접 with transaction.atomic():으로 감싸야 안전.

4) 시그널 폭발 #

대량 데이터를 건드릴 때 post_save 시그널이 매 행마다 외부 호출을 트리거할 수 있습니다. 마이그레이션성 작업에서는 시그널 임시 disconnect 또는 bulk_update 사용 (#5 참고).

정리 #

이번 글에서 잡은 것:

  • manage.py는 execute_from_command_line의 진입점
  • 위치: myapp/management/commands/<name>.py, __init__.py 둘 다 필수
  • BaseCommand + handle(*args, **options), add_arguments(parser)
  • 인자: positional, optional, flag (action="store_true"), nargs="+"
  • 출력: self.stdout.write + self.style.SUCCESS/ERROR/...
  • 종료: CommandError (코드 1, stderr)
  • 자주 쓰는 부류: 만료 정리, 통계 적재 (update_or_create), 데이터 마이그레이션 (iterator(chunk_size=...))
  • call_command(...)로 코드/다른 커맨드에서 호출
  • cron / systemd timer와 결합 — 절대경로, 로그 리다이렉트
  • 테스트: StringIO + call_command(..., stdout=out)
  • 함정: __init__.py 누락, 자동 트랜잭션 없음, 시그널 폭발

다음 글(#3 쿼리 최적화)에서는 중급 #2의 ORM 위에 본격적인 N+1 진단과 해결, select_related/prefetch_related, 인덱스, bulk_* 까지 — 운영에서 가장 자주 마주치는 성능 병목들의 도구상자를 정리합니다.

X