Django上級 #2 Custom management commands

フレームワーク Django management commands
読了 8分

#1 Async views で非同期陣営を見たなら、今回は運用のもう 1 つの軸である コマンドラインの作業 です。manage.py の一語で回すバッチ作業、データマイグレーション、メンテナンススクリプトの出番です。

manage.py の正体 #

Django プロジェクトを作ると自動で付いてくる 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 が Django の慣例です。

実践 — 期限切れトークンの掃除 #

運用で最もよく作る部類。

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 が一度失敗してもう一度回しても重複が生じません。

実践 — データマイグレーション #

スキーマはそのままにして データだけ を変えなければならないことが多いです (例: ユーザー名の空白整理、カテゴリ統合など)。2 つの道:

  1. Django マイグレーションの 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 が貧弱。Python パスを明示
  • プロジェクトディレクトリへ cd — 相対 import が壊れない
  • stdout/stderr の両方をロギング — >> ... 2>&1
  • 環境変数が必要なら BASH_ENV=/etc/profile または cron ファイルの上部で直接 export

systemd timer #

最近の Linux は 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 の漏れが最も多いです。両方とも空ファイルでもあれば OK。

また アプリが 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