在线 -
搜索 AIGC编程网
← 返回题目列表

Django 的自定义管理命令怎么写？有哪些实践要点？

简单 第 22 / 27 题 更新于 2026/08/02
Djangomanagement command定时任务脚本

简化版

自定义管理命令就是「能用 python manage.py <你的命令> 跑起来的脚本」，写法是在 app 下建 management/commands/<命令名>.py（两层目录都必须有 __init__.py），定义一个继承 BaseCommand 的 Command 类，把逻辑写在 handle(self, *args, **options) 里。为什么不直接写一个独立的 .py 脚本？因为管理命令会自动帮你做好 Django 环境初始化——settings 加载、app registry 就绪、数据库连接建立，你直接 import 模型就能用；而独立脚本得自己 os.environ.setdefault("DJANGO_SETTINGS_MODULE", ...) 加 django.setup()，还容易踩「app 尚未加载」的坑。参数用 add_arguments(self, parser) 声明（底层就是 argparse），支持位置参数、--flag、类型转换和默认值。输出要用 self.stdout.write() 而不是 print——因为它可以被测试重定向、支持 self.style.SUCCESS/WARNING/ERROR 着色，并且遵守 --verbosity 级别。几个实践要点：出错时 raise CommandError("...")（Django 会打印红色错误并以非零退出码结束，而不是甩一整页 traceback）、需要事务包裹就设 @transaction.atomic 或类属性 output_transaction、长任务要打印进度并支持中断续跑、定时任务用 cron/Celery Beat 调用它。测试也很方便：call_command("mycmd", "--dry-run", stdout=out) 可以直接在测试里调用并捕获输出。核心记忆：management/commands/xxx.py + Command(BaseCommand) + handle()；参数靠 add_arguments；输出用 self.stdout.write 配 self.style；报错用 CommandError。

详细版

管理命令 vs 其他方案：

方案Django 环境参数解析可测试适用
管理命令✅ 自动就绪✅ argparse✅ call_command绝大多数脚本任务
独立 .py 脚本❌ 要手写 django.setup()自己写难不推荐
manage.py shell -c✅❌❌临时一次性操作
Celery 任务✅参数即函数签名✅需要异步/分布式
# ★目录结构（两层都要 __init__.py）★
# myapp/
#   management/
#     __init__.py          ← ★必须★
#     commands/
#       __init__.py        ← ★必须★
#       cleanup_articles.py

from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from django.utils import timezone

class Command(BaseCommand):
    help = "清理 N 天前的草稿文章"           # ★python manage.py help cleanup_articles★

    def add_arguments(self, parser):        # ★parser 就是 argparse.ArgumentParser★
        parser.add_argument("days", type=int, help="保留最近多少天")   # ★位置参数★
        parser.add_argument("--dry-run", action="store_true",
                            help="只打印不执行")                       # ★开关★
        parser.add_argument("--batch-size", type=int, default=1000)
        parser.add_argument("--status", choices=["draft", "archived"],
                            default="draft")

    def handle(self, *args, **options):     # ★核心逻辑★
        days = options["days"]
        dry_run = options["dry_run"]        # ★注意：--dry-run → dry_run（横线变下划线）★
        verbosity = options["verbosity"]    # ★内置参数 0/1/2/3★

        if days < 1:
            raise CommandError("days 必须 ≥ 1")     # ★非零退出码 + 红色输出★

        cutoff = timezone.now() - timezone.timedelta(days=days)
        qs = Article.objects.filter(status=options["status"], created__lt=cutoff)
        total = qs.count()

        if verbosity >= 1:
            self.stdout.write(f"待清理 {total} 条（cutoff={cutoff:%Y-%m-%d}）")

        if dry_run:
            self.stdout.write(self.style.WARNING("dry-run，未实际删除"))
            return

        deleted = 0
        while True:
            ids = list(qs.values_list("pk", flat=True)[:options["batch_size"]])
            if not ids:
                break
            with transaction.atomic():                # ★每批一个事务★
                n, _ = Article.objects.filter(pk__in=ids).delete()
            deleted += n
            self.stdout.write(f"  进度 {deleted}/{total}")

        self.stdout.write(self.style.SUCCESS(f"完成，删除 {deleted} 条"))

# ★调用方式★
# python manage.py cleanup_articles 30 --dry-run
# python manage.py cleanup_articles 30 --batch-size 500 -v 2

# ★在代码/测试里调用★
from django.core.management import call_command
from io import StringIO
out = StringIO()
call_command("cleanup_articles", 30, dry_run=True, stdout=out)   # ★参数名用下划线★
assert "dry-run" in out.getvalue()

⚠️ 三个必须记住的点：① 目录结构不能错：必须是 <app>/management/commands/<命令名>.py，而且 management/ 和 commands/ 两层都要有 __init__.py（否则 Django 找不到，manage.py 里看不见你的命令，而且不会有任何报错提示——这是新手最常卡住的地方）。命令名就是文件名（去掉 .py），下划线会保留（clean_up.py → manage.py clean_up），文件名以下划线开头的会被忽略。app 还必须在 INSTALLED_APPS 里。② 输出一律用 self.stdout.write()，不要用 print。原因有三：self.stdout 可以被 call_command(stdout=...) 重定向，让命令变得可测试；配合 self.style.SUCCESS/WARNING/ERROR/NOTICE 能自动着色（并在非 TTY 环境下自动关闭颜色）；还能配合 options["verbosity"]（-v 0/1/2/3）做输出分级。注意 self.stdout.write() 默认会自动加换行（和 print 一样），要抑制就传 ending=""。③ 出错用 raise CommandError(...) 而不是 sys.exit() 或裸抛异常。CommandError 会被 Django 捕获，打印成简洁的红色错误信息并以退出码 1 结束——这对 cron 和 CI 很重要（它们靠退出码判断成败）；而裸抛异常会甩出一整页 traceback，sys.exit() 则会跳过 Django 的清理逻辑。

完整版教学

一、为什么要用管理命令而不是独立脚本

★ 独立脚本要自己做的事：
  # my_script.py
  import os, django
  os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
  django.setup()                        # ★必须在 import 模型之前★
  from myapp.models import Article      # ★放到 setup() 之后，否则 AppRegistryNotReady★
  ...
  ✗ ★import 顺序稍错就报 AppRegistryNotReady★
  ✗ 换 settings（dev/prod）要改代码或加环境变量
  ✗ 参数解析要自己写
  ✗ 数据库连接的开启/关闭要自己管
  ✗ 没有统一的入口，运维不知道有哪些脚本

★ 管理命令自动帮你做的：
  ┌────────────────────────────────────────────────┐
  │ ① ★加载 settings★（--settings / DJANGO_SETTINGS_MODULE）│
  │ ② ★django.setup()：app registry 就绪★           │
  │ ③ ★argparse 解析参数★（含一批内置参数）          │
  │ ④ ★系统检查（system checks）★                    │
  │ ⑤ ★执行后关闭数据库连接★                        │
  │ ⑥ ★统一入口：manage.py help 能列出所有命令★      │
  └────────────────────────────────────────────────┘

★ 内置参数（★所有命令都自动有★）：
  --settings=myproject.settings.prod    # ★切换配置★
  --pythonpath=/path/to/project
  -v 0/1/2/3  (--verbosity)             # ★输出详细程度★
  --traceback                            # ★出错时打印完整栈★
  --no-color / --force-color             # 颜色控制
  --skip-checks                          # ★跳过系统检查（提速）★

★ ★运维视角的价值★：
  python manage.py help                  # ★列出所有可用命令（含自定义的）★
  → 新同事一眼看到项目有哪些运维操作
  → 对比"scripts/ 目录下 30 个来路不明的 .py"

★ 什么时候还是该用别的：
  ┌──────────────────────────┬──────────────────────────┐
  │ ★一次性的临时操作★        │ manage.py shell / shell_plus│
  │ ★需要异步/重试/分布式★    │ ★Celery 任务★             │
  │ ★需要 Web 触发★          │ 视图 + Celery             │
  │ ★纯数据修正（一次性）★    │ ★数据迁移（RunPython）★   │
  └──────────────────────────┴──────────────────────────┘
  ★ 关键区分：★管理命令是"可重复执行的运维操作"★，
    ★数据迁移是"只执行一次的数据修正"★ —— 别混用

管理命令相比独立脚本的核心价值是「Django 环境自动就绪」——独立脚本要自己 django.setup()，而且 import 顺序稍错就会报 AppRegistryNotReady（模型必须在 setup() 之后导入），还得自己处理 settings 切换、参数解析、数据库连接管理。管理命令自动完成这六件事，并且提供了一批所有命令都有的内置参数：--settings（切换配置）、-v（输出级别）、--traceback（打印完整栈）、--skip-checks（跳过系统检查提速）。从运维视角看还有个隐藏价值：manage.py help 能列出所有命令，新同事一眼就知道项目有哪些运维操作——对比 scripts/ 目录下三十个来路不明的 .py 文件，差别很明显。还要分清边界：管理命令是「可重复执行的运维操作」，数据迁移的 RunPython 是「只执行一次的数据修正」，两者别混用。

二、参数设计与 argparse

★ add_arguments 拿到的就是 argparse.ArgumentParser：
  def add_arguments(self, parser):
      # ★位置参数（必填）★
      parser.add_argument("user_id", type=int)
      parser.add_argument("emails", nargs="+")            # ★1 个或多个★
      parser.add_argument("files", nargs="*")             # ★0 个或多个★

      # ★可选参数★
      parser.add_argument("--days", type=int, default=7)
      parser.add_argument("--dry-run", action="store_true")     # ★布尔开关★
      parser.add_argument("--no-cache", action="store_false", dest="use_cache")
      parser.add_argument("--tag", action="append")             # ★可重复：--tag a --tag b★
      parser.add_argument("--mode", choices=["fast", "safe"], default="safe")
      parser.add_argument("-f", "--force", action="store_true") # ★短选项★
      parser.add_argument("--since", type=lambda s: datetime.strptime(s, "%Y-%m-%d"))

      # ★互斥组★
      group = parser.add_mutually_exclusive_group()
      group.add_argument("--all", action="store_true")
      group.add_argument("--ids", nargs="+", type=int)

★ ★命名转换规则（★容易搞错★）★：
  --dry-run          → options["dry_run"]        ★横线变下划线★
  --batch-size       → options["batch_size"]
  dest="use_cache"   → options["use_cache"]      ★显式指定 dest★
  位置参数 user_id   → options["user_id"]

★ handle 的签名：
  def handle(self, *args, **options):
      # ★现代 Django 里位置参数也在 options 里★，args 一般是空的
      user_id = options["user_id"]
  ★ 老代码里 def handle(self, *args, **options) 里用 args[0] 是 optparse 时代的写法

★ ★参数设计的实践建议★：
  ① ★危险操作默认安全★：
     ✓ --dry-run 默认 True？不 —— ★更常见的是提供 --dry-run 让人主动预演★
     ✓ 真正危险的（删数据）加 ★--yes-i-am-sure★ 或交互确认
  ② ★给出 help 文本★：cron 里的人半年后看不懂 --mode=2
  ③ ★参数校验放前面★：先校验全部参数再动手，别做到一半才报错
  ④ ★提供 --limit 便于试跑★：先跑 100 条看效果
  ⑤ ★时间参数用 ISO 格式★并明确时区

★ 交互确认：
  def handle(self, *args, **options):
      if not options["force"]:
          answer = input("将删除 10000 条数据，确认？[y/N] ")
          if answer.lower() != "y":
              self.stdout.write("已取消")
              return
  ★ 注意：★cron 里没有 stdin★ → 必须提供 --force 之类的旁路
  ★ 内置命令的做法：flush 命令用 --no-input（★沿用这个命名更一致★）

add_arguments 拿到的就是标准的 argparse.ArgumentParser，所以 argparse 的全部能力都可用：nargs、action="store_true"/"append"、choices、type= 自定义转换函数、短选项、互斥组。有个容易搞错的命名转换规则：--dry-run 在 options 里是 dry_run（横线变下划线）。参数设计上有五条实践建议：危险操作提供 --dry-run 让人主动预演（真正危险的删数据操作再加 --force 或交互确认）、认真写 help 文本（半年后看 cron 里的 --mode=2 谁也看不懂）、参数校验全部放在动手之前（别做到一半才报错）、提供 --limit 便于试跑、时间参数用 ISO 格式并明确时区。交互确认要注意 cron 里没有 stdin，必须提供 --force/--no-input 之类的旁路（沿用 Django 内置命令的 --no-input 命名更一致）。

三、输出、日志与进度

★ 三种输出方式的定位：
  ┌──────────────────┬──────────────────────────────────────┐
  │ ★self.stdout★    │ ★给"人"看的运行反馈★（可重定向、可着色）│
  │ ★logging★        │ ★给"日志系统"的记录★（有级别、有上下文）│
  │ print            │ ★不要用★                              │
  └──────────────────┴──────────────────────────────────────┘

★ self.stdout / self.stderr：
  self.stdout.write("普通信息")
  self.stdout.write(self.style.SUCCESS("成功"))      # ★绿★
  self.stdout.write(self.style.WARNING("警告"))      # ★黄★
  self.stdout.write(self.style.ERROR("错误"))        # ★红★
  self.stdout.write(self.style.NOTICE("提示"))
  self.stdout.write("不换行", ending="")             # ★默认会加 \n★
  self.stderr.write(self.style.ERROR("错误到 stderr"))
  ★ style 在★非 TTY（重定向到文件、cron）时自动关闭颜色★
  ★ 也可以 --no-color 强制关闭

★ ★verbosity 分级输出（内置参数）★：
  def handle(self, *args, **options):
      v = options["verbosity"]      # 0=静默 1=正常(默认) 2=详细 3=非常详细
      if v >= 2:
          self.stdout.write(f"处理 {obj.pk}: {obj}")
      if v >= 1:
          self.stdout.write(self.style.SUCCESS("完成"))
  ★ cron 里用 -v 0 只在出错时有输出（★"没有消息就是好消息"★）

★ ★同时接 logging（生产必备）★：
  import logging
  logger = logging.getLogger(__name__)

  def handle(self, *args, **options):
      logger.info("cleanup started days=%s", options["days"])
      try:
          ...
      except Exception:
          logger.exception("cleanup failed")        # ★带 traceback 进日志★
          raise CommandError("清理失败，详见日志")
  ★ 为什么两套都要：
    stdout → ★人工执行时的即时反馈★
    logging → ★cron/容器里的结构化记录、能接 Sentry★

★ ★进度显示（长任务必备）★：
  total = qs.count()
  t0 = time.perf_counter()
  for i, obj in enumerate(qs.iterator(chunk_size=1000), 1):
      process(obj)
      if i % 1000 == 0:
          elapsed = time.perf_counter() - t0
          rate = i / elapsed
          eta = (total - i) / rate
          self.stdout.write(f"  {i}/{total} ({i/total:.0%}) "
                            f"rate={rate:.0f}/s eta={eta:.0f}s")
  ★ 至少要能回答：★进度多少、速率多少、还要多久★
  ★ 第三方：tqdm（★但 cron 日志里会很难看，建议只在 TTY 时启用★）
    from tqdm import tqdm
    it = tqdm(qs.iterator()) if sys.stdout.isatty() else qs.iterator()

★ 退出码（★cron 和 CI 靠它判断成败★）：
  正常结束           → ★0★
  raise CommandError → ★1★（★推荐★，输出简洁）
  未捕获的异常       → ★1 + 完整 traceback★
  sys.exit(2)        → 2（★可用于区分错误类型★，但跳过了 Django 清理）

三种输出方式定位不同：self.stdout 是给「人」看的运行反馈（可重定向、可着色）、logging 是给日志系统的记录（有级别、能接 Sentry）、print 不要用。生产环境两套都要——人工执行时看 stdout 的即时反馈，cron/容器里看结构化日志。self.style 在非 TTY 环境（重定向到文件、cron）会自动关闭颜色，不用担心日志里出现乱码转义符。verbosity 是所有命令都有的内置参数，用它做输出分级，cron 里配 -v 0 就能实现「没有消息就是好消息」。长任务必须显示进度——至少能回答「进度多少、速率多少、还要多久」；用 tqdm 的话建议只在 sys.stdout.isatty() 为真时启用，否则 cron 日志会被进度条刷屏。最后退出码很关键：cron 和 CI 靠它判断成败，所以出错一定要走 CommandError（退出码 1、输出简洁）而不是静默 return。

四、事务、幂等与可恢复

★ 事务的三种粒度：
  ① ★整个命令一个事务★（数据量小、要求原子）
     class Command(BaseCommand):
         @transaction.atomic
         def handle(self, *args, **options): ...
     ✗ 大数据量时★长事务持锁、失败全回滚★

  ② ★分批各自事务（★推荐★）★
     for batch in chunks(ids, 1000):
         with transaction.atomic():
             process(batch)
     ✓ 失败只丢一批、锁竞争小、可断点续传

  ③ ★不用事务★（只读命令、生成报表）

★ ★output_transaction 属性★（容易误解）：
  class Command(BaseCommand):
      output_transaction = True
  ★ 它★不是"包一个事务"★！
  ★ 它只是让命令★输出的 SQL★（如 sqlmigrate）被 BEGIN/COMMIT 包裹
  → ★只对"输出 SQL 文本"的命令有意义★

★ ★幂等性：命令应该能安全地重复执行★
  为什么重要：
    - cron 可能重复触发（★上次没跑完，这次又来了★）
    - 失败重试
    - 人工手动补跑
  怎么做：
    ✓ ★用 update_or_create / get_or_create★
    ✓ ★bulk_create(update_conflicts=True)★（upsert）
    ✓ ★先检查状态再操作★（already_processed 标记）
    ✗ 危险：直接 create()（★重复执行产生重复数据★）
    ✗ 危险：F("count") + 1（★重复执行会多加★）

★ ★防重入（并发保护）★：
  ✗ 上一次还没跑完，cron 又启动了一个 → ★两个进程同时改同一批数据★
  ✓ 方案一：★文件锁★
    import fcntl                              # Unix
    with open("/tmp/mycmd.lock", "w") as f:
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            self.stdout.write("已有实例在运行，跳过"); return
  ✓ 方案二：★缓存锁（分布式）★
    if not cache.add("lock:mycmd", "1", timeout=3600):    # ★原子★
        return
    try: ...
    finally: cache.delete("lock:mycmd")
  ✓ 方案三：★数据库行锁★
    with transaction.atomic():
        job = Job.objects.select_for_update(nowait=True).get(name="mycmd")

★ ★断点续传★：
  # 记录进度
  state, _ = CommandState.objects.get_or_create(name="import")
  qs = Article.objects.filter(pk__gt=state.last_pk).order_by("pk")
  for batch in chunked(qs, 1000):
      with transaction.atomic():
          process(batch)
          state.last_pk = batch[-1].pk
          state.save(update_fields=["last_pk"])
  ★ ★跑超过 10 分钟的命令都应该能续跑★

★ ★优雅中断（Ctrl+C / SIGTERM）★：
  import signal
  class Command(BaseCommand):
      def handle(self, *a, **kw):
          self._stop = False
          signal.signal(signal.SIGTERM, self._on_signal)
          signal.signal(signal.SIGINT, self._on_signal)
          for batch in ...:
              if self._stop:
                  self.stdout.write(self.style.WARNING("收到中断，安全退出"))
                  break
              process(batch)
      def _on_signal(self, signum, frame):
          self._stop = True                 # ★不立即退出，跑完当前批★
  ★ 容器化部署时 ★K8s 发 SIGTERM 后有宽限期★ —— 用好它能避免数据半途而废

事务粒度的选择和批量操作一样：分批各自事务是推荐做法（失败只丢一批、锁竞争小、可断点续传）。这里要澄清一个常见误解：output_transaction = True 不是「给命令包一个事务」，它只是让命令输出的 SQL 文本（如 sqlmigrate）被 BEGIN/COMMIT 包裹，只对生成 SQL 的命令有意义。幂等性是管理命令的核心要求——因为 cron 可能重复触发、失败要重试、人工要补跑，所以要用 update_or_create、bulk_create(update_conflicts=True) 或状态标记，避免裸 create() 和 F("count") + 1 这类重复执行就出错的写法。防重入有三种方案（文件锁、缓存锁 cache.add 的原子性、数据库行锁 select_for_update(nowait=True)），分布式部署选缓存锁。跑超过 10 分钟的命令都应该能断点续传。最后优雅中断在容器化部署下很实用：捕获 SIGTERM 后只置标志位、跑完当前批再退出，配合 K8s 的宽限期就能避免数据处理到一半被强杀。

五、测试与调度

★ ★测试管理命令（比想象中容易）★：
  from django.core.management import call_command
  from io import StringIO

  class CleanupTests(TestCase):
      def test_dry_run_does_not_delete(self):
          ArticleFactory.create_batch(5, status="draft",
                                      created=timezone.now() - timedelta(days=60))
          out = StringIO()
          call_command("cleanup_articles", 30, dry_run=True, stdout=out)
          self.assertEqual(Article.objects.count(), 5)      # ★没删★
          self.assertIn("dry-run", out.getvalue())

      def test_actually_deletes(self):
          ...
          call_command("cleanup_articles", 30, stdout=StringIO())
          self.assertEqual(Article.objects.count(), 0)

      def test_invalid_days(self):
          with self.assertRaises(CommandError):
              call_command("cleanup_articles", 0)

  ★ 关键点：
    - ★call_command 的参数名用下划线★（dry_run 不是 dry-run）
    - ★位置参数直接传★，可选参数用关键字
    - ★stdout=StringIO() 捕获输出★（这就是"别用 print"的理由）
    - ★也可以传字符串形式★：call_command("cmd", "--dry-run")

★ ★调度方式对比★：
  ┌──────────────────┬──────────────────────────────────────┐
  │ ★crontab★        │ ★最简单★；要处理路径/虚拟环境/日志     │
  │ ★Celery Beat★    │ ★已有 Celery 时首选★；有重试和监控     │
  │ ★K8s CronJob★    │ ★容器化部署的标准做法★                │
  │ systemd timer    │ 比 cron 更好的日志和依赖管理           │
  │ django-crontab   │ 把调度写进 settings（★小项目方便★）    │
  └──────────────────┴──────────────────────────────────────┘

★ crontab 的正确写法（★踩坑集中区★）：
  # ✗ 常见错误
  0 3 * * * python manage.py cleanup 30
  #   ★① python 不是虚拟环境里的 ② 工作目录不对 ③ 没有日志 ④ 环境变量缺失★

  # ✓ 正确
  0 3 * * * cd /app && /app/.venv/bin/python manage.py cleanup 30 \
            --settings=myproject.settings.prod -v 1 \
            >> /var/log/cleanup.log 2>&1
  ★ 要点：★绝对路径★、★cd 到项目目录★、★重定向日志（含 stderr）★、
        ★显式 --settings★（cron 环境没有你 shell 里的 export）

★ K8s CronJob：
  apiVersion: batch/v1
  kind: CronJob
  spec:
    schedule: "0 3 * * *"
    concurrencyPolicy: ★Forbid★              # ★防重入★
    jobTemplate:
      spec:
        backoffLimit: 2                       # 重试次数
        template:
          spec:
            containers:
            - name: cleanup
              command: ["python", "manage.py", "cleanup_articles", "30"]
  ★ concurrencyPolicy: Forbid ★天然解决了防重入★

★ Celery Beat 调 management command：
  from django.core.management import call_command
  @shared_task
  def cleanup_task():
      call_command("cleanup_articles", 30)
  ★ 好处：★复用同一份逻辑★，既能手动跑也能定时跑
  ★ 也可以直接用 django-celery-beat 在 admin 里配置周期

★ ★监控（生产必备）★：
  - ★成功/失败要能被发现★（退出码 + 告警）
  - ★死信监控★：用 healthchecks.io / Dead man's snitch 这类
    "该来的心跳没来就告警"的服务
  - 记录到数据库：CommandRun(name, started, finished, status, message)

测试管理命令比想象中容易：call_command("cleanup_articles", 30, dry_run=True, stdout=out) 就能在测试里跑起来并捕获输出——注意参数名要用下划线（dry_run 而非 dry-run），位置参数直接传、可选参数用关键字；这也正是「别用 print」的实际理由。调度方式上：已有 Celery 就用 Celery Beat（有重试和监控）、容器化部署用 K8s CronJob（concurrencyPolicy: Forbid 天然解决了防重入）、传统部署用 crontab。crontab 是踩坑集中区——必须用虚拟环境的绝对路径 python、先 cd 到项目目录、重定向日志且带上 2>&1、显式 --settings（cron 环境没有你 shell 里 export 的变量）。Celery Beat 里推荐用 call_command 包一层，这样同一份逻辑既能手动跑也能定时跑。最后监控是生产必备：至少要保证失败能被发现，更好的是用「该来的心跳没来就告警」的死信监控服务。

六、常用内置命令与扩展技巧

★ 值得知道的内置命令：
  ┌────────────────────────────┬────────────────────────────┐
  │ ★makemigrations --check★    │ ★CI 里检查有无漏生成的迁移★ │
  │ ★migrate --plan★            │ 预览将执行哪些迁移          │
  │ ★sqlmigrate app 0003★       │ ★查看迁移对应的 SQL★       │
  │ ★check --deploy★            │ ★上线前的安全配置检查★     │
  │ ★shell -c "code"★           │ 一次性执行代码              │
  │ ★dbshell★                   │ 进数据库命令行              │
  │ ★showmigrations★            │ 看迁移状态                  │
  │ ★clearsessions★             │ ★清理过期 session（该定时跑）★│
  │ ★collectstatic★             │ 收集静态文件                │
  │ ★createsuperuser --noinput★ │ 脚本化创建管理员            │
  │ ★diffsettings★              │ 看与默认配置的差异          │
  └────────────────────────────┴────────────────────────────┘

★ 其他 BaseCommand 属性：
  class Command(BaseCommand):
      help = "..."                     # ★帮助文本★
      requires_migrations_checks = True   # ★运行前检查有无未应用的迁移★
      requires_system_checks = ["database"]  # 或 [] 跳过（★提速★）
      suppressed_base_arguments = {"--verbosity"}   # ★隐藏内置参数★

★ ★复用逻辑：命令只做"入口"★（★重要的设计原则★）
  ✗ 把 300 行业务逻辑写在 handle 里
     → 无法被视图/Celery/其他命令复用；难测试
  ✓ 逻辑放 services，命令只做参数解析和输出：
     # services/cleanup.py
     def cleanup_articles(days, dry_run=False, on_progress=None): ...
     # management/commands/cleanup_articles.py
     def handle(self, *a, **opts):
         n = cleanup_articles(opts["days"], opts["dry_run"],
                              on_progress=lambda i, t: self.stdout.write(...))
         self.stdout.write(self.style.SUCCESS(f"完成 {n}"))
  ★ 好处：★同一逻辑能被命令、Celery 任务、admin action、API 复用★

★ 覆盖 Django 内置命令：
  在自己 app 里建同名文件（如 management/commands/runserver.py）
  ★ 该 app 要排在 INSTALLED_APPS 中 django.contrib.staticfiles 之前才生效
  ★ 谨慎使用 —— ★会让新同事困惑"为什么 runserver 行为不一样"★

★ AppCommand / LabelCommand（少用但知道）：
  class Command(AppCommand):
      def handle_app_config(self, app_config, **options): ...   # ★每个 app 调一次★
  class Command(LabelCommand):
      def handle_label(self, label, **options): ...             # ★每个参数调一次★

★ 一句话总结：
  ★"管理命令 = Django 环境就绪的脚本入口；
    参数交给 add_arguments、输出交给 self.stdout + style、
    错误交给 CommandError、业务逻辑交给 services 层；
    定时跑要保证幂等、防重入、可续跑、有监控。"★

有几个内置命令值得知道：makemigrations --check（CI 里检查有无漏生成的迁移）、check --deploy（上线前的安全配置检查）、sqlmigrate（查看迁移对应的 SQL）、clearsessions（清理过期 session，这个应该定时跑但很多项目忘了）、diffsettings。BaseCommand 还有几个实用属性：requires_migrations_checks（运行前检查有无未应用的迁移）、requires_system_checks = []（跳过检查提速）、suppressed_base_arguments（隐藏内置参数）。最重要的设计原则是「命令只做入口」——把 300 行业务逻辑写在 handle 里会导致它无法被视图、Celery 任务、admin action 复用，也难以测试；正确做法是逻辑放 services 层，命令只负责参数解析和输出（进度回调用 on_progress 之类的钩子传进去）。最后，虽然可以通过同名文件覆盖 Django 内置命令，但要谨慎——它会让新同事困惑「为什么这个项目的 runserver 行为不一样」。

记忆钩子：「管理命令＝★Django 环境自动就绪的脚本入口★。三件事定形态：★<app>/management/commands/<名>.py（两层目录都必须有 __init__.py，缺了就静默找不到——新手最常卡的地方）★ + ★class Command(BaseCommand)★ + ★handle(self, *args, **options)★。相比独立脚本，它自动做了：加载 settings、★django.setup() 让 app registry 就绪（独立脚本里模型必须在 setup() 之后 import，否则 AppRegistryNotReady）★、argparse 解析、系统检查、关连接，还给了一批内置参数（★—settings / -v / —traceback / —skip-checks★）。三条硬规矩：★① 参数用 add_arguments(parser)（就是 argparse），注意 —dry-run 在 options 里叫 dry_run（横线变下划线）★；★② 输出用 self.stdout.write() 而不是 print★——因为它★可被 call_command(stdout=…) 重定向所以可测试★、支持 ★self.style.SUCCESS/WARNING/ERROR★ 且非 TTY 时自动关色、能配合 verbosity 分级；★③ 出错 raise CommandError★（简洁红色 + ★退出码 1★，cron 和 CI 靠退出码判断成败），别裸抛异常或 sys.exit。生产四要素：★幂等★（cron 会重复触发 → 用 update_or_create / update_conflicts，★别裸 create() 或 F()+1★）、★防重入★（文件锁 / ★cache.add 原子锁★ / select_for_update(nowait) / ★K8s 的 concurrencyPolicy: Forbid★）、★断点续传★（超 10 分钟的命令都该能续跑）、★监控★（退出码告警 + 死信心跳）。事务用★分批各自 atomic★而不是整个命令一个大事务；★注意 output_transaction 不是『包事务』★，它只让输出的 SQL 文本被 BEGIN/COMMIT 包住。★最重要的设计原则：命令只做入口，业务逻辑放 services 层★——这样同一逻辑能被命令、Celery、admin action、API 复用。crontab 四个坑：★虚拟环境绝对路径 + cd 到项目目录 + 重定向日志带 2>&1 + 显式 —settings★。测试用 ★call_command(‘cmd’, 30, dry_run=True, stdout=StringIO())★（★参数名用下划线★）。」

七、常见误区与追问

  • 误区：把脚本文件放进 app 目录下就能被 manage.py 找到。 路径必须精确是 <app>/management/commands/<命令名>.py，而且 management/ 和 commands/ 两层目录都必须有 __init__.py（即使在 Python 3 的隐式命名空间包时代，Django 的命令发现机制仍然依赖它）。更麻烦的是缺了 __init__.py 不会有任何报错——命令就是不出现在 manage.py help 的列表里，你会以为是代码写错了。检查清单：① 两层 __init__.py 都在吗？② app 在 INSTALLED_APPS 里吗？③ 文件名是不是以下划线开头（会被忽略）？④ 类名是不是恰好叫 Command？ 另外命令名就是文件名去掉 .py，下划线会原样保留（clean_up.py → manage.py clean_up）。
  • 误区：在管理命令里用 print 输出没什么问题，反正都是打到控制台。 有三个实际损失。① 不可测试：call_command(..., stdout=out) 只能捕获 self.stdout 的内容，print 会直接打到真正的标准输出，测试里既拿不到也断言不了——这是「用 self.stdout」最实际的理由。② 没有着色和分级：self.style.SUCCESS/WARNING/ERROR 能让关键信息在一大堆日志里凸显出来，而且在非 TTY 环境（重定向到文件、cron、容器日志）会自动关闭颜色，不会留下一堆 ANSI 转义符；print 要么没颜色，要么手写转义符污染日志。③ 不遵守 --verbosity：self.stdout 配合 options["verbosity"] 能实现「cron 里 -v 0 静默、排查时 -v 3 详细」，而 print 永远都打。此外错误信息应该写到 self.stderr，这样重定向 stdout 时错误仍然可见。
  • 误区：命令出错时直接 return 或者让异常抛出去都一样。 差别在退出码和可读性，而这两点对自动化调度至关重要。静默 return 的退出码是 0——cron 和 CI 会认为「成功执行」，于是失败被完全掩盖（这是最危险的一种，你可能几周后才发现某个清理任务其实一直在报错跳过）。裸抛异常退出码是 1（正确），但会打出一整页 traceback，在 cron 邮件或容器日志里非常难读。raise CommandError("清理失败：xxx") 是 Django 为此专门设计的：它会被顶层捕获，输出一行简洁的红色错误信息，以退出码 1 结束——既能被自动化正确识别，又对人友好；需要看完整栈时加 --traceback 即可。至于 sys.exit()，它会跳过 Django 的连接清理逻辑，只在需要用不同退出码区分错误类型时才考虑。
  • 误区：定时任务不用考虑重复执行，cron 每小时只会跑一次。 现实中重复执行的路径至少有四条：① 上一次还没跑完，下一次就到点了（任务变慢是常态，比如数据量增长后从 20 分钟变成 70 分钟）；② 失败后人工补跑；③ 容器重启或 K8s 重新调度导致重试；④ 部署了多个实例，每个都装了 crontab。所以幂等性是管理命令的基本要求：用 update_or_create/get_or_create 而不是裸 create()（否则重复执行产生重复数据）、用 bulk_create(update_conflicts=True) 做 upsert、用状态字段标记「已处理」；特别要警惕 F("count") + 1 这类累加操作——重复执行会多加。同时要加防重入保护：单机用文件锁（fcntl.flock + LOCK_NB）、分布式用缓存锁（cache.add() 的原子性，注意设 timeout 防死锁）或数据库行锁（select_for_update(nowait=True)）、K8s 直接配 concurrencyPolicy: Forbid。
  • 误区：管理命令和数据迁移里的 RunPython 差不多，用哪个都行。 两者的定位完全不同。数据迁移（RunPython）是「只执行一次、随代码版本走」的数据修正——它记录在 django_migrations 表里，每个环境恰好执行一次，且执行顺序与其他迁移严格绑定（比如「加了字段之后立刻填充默认值」）。管理命令是「可反复执行的运维操作」——清理、导入、重算、对账，什么时候跑由你决定。误用的后果很实际：把一次性的数据修正写成管理命令，容易忘记在某个环境执行（或者不知道执行过没有）；反过来把常规运维操作写进迁移，会导致迁移变慢、无法重跑、回滚困难，而且在 migrate 时执行大批量数据操作可能造成部署卡死。还有个折中做法：在迁移的 RunPython 里调用 services 层的函数，同一份逻辑再包一个管理命令供以后手动重跑。
  • 追问：怎么在管理命令里显示进度，同时又不污染 cron 日志？ 关键是区分 TTY 和非 TTY 环境。人工执行时（TTY）用 tqdm 之类的进度条最直观，但它靠 \r 回车刷新同一行——重定向到文件后会变成成千上万行垃圾。做法是判断 sys.stdout.isatty()：it = tqdm(qs.iterator()) if sys.stdout.isatty() else qs.iterator()。非 TTY 环境下改用周期性打印（每 1000 条或每 30 秒一行），内容包含已处理数 / 总数 / 百分比 / 速率 / 预计剩余时间——这几项能回答运维最关心的问题「它卡住了吗？还要多久？」。配合 options["verbosity"] 分级：-v 0 完全静默（只在出错时输出）、-v 1 打里程碑、-v 2 打每批、-v 3 打每条。另外总数用 count() 要小心——大表上这一条查询本身就很慢，可以改成估算或干脆不显示百分比。
  • 追问：为什么建议「命令只做入口，业务逻辑放 services 层」？ 因为同一段业务逻辑往往需要多种触发方式。举例：「重算某个用户的积分」这件事，可能需要在管理命令里批量跑（数据修复）、在 Celery 任务里异步跑（用户操作后触发）、在 admin action 里点按钮跑（客服手动处理）、在 API 里同步跑（前端请求）。如果逻辑写死在 handle() 里，后面三种场景要么复制代码、要么用 call_command() 硬凑（参数只能是字符串、输出只能靠捕获 stdout、异常处理很别扭）。抽到 services/points.py 的普通函数后，四个入口都只是薄薄一层。附带的好处还有：函数比命令好测（不用 call_command、不用捕获输出、能直接断言返回值）、进度和输出解耦（用 on_progress 回调传进去，命令里打 stdout、Celery 里更新任务状态、API 里推 WebSocket）。判断标准很简单：handle() 超过 50 行就该考虑抽出去了。
  • 追问：管理命令在容器化部署（K8s）下有哪些特别要注意的？ 五点。① 用 CronJob 而不是在容器里装 crontab——CronJob 有原生的调度、重试（backoffLimit）、历史记录，而容器里的 crontab 需要额外的进程管理器且日志难收集。② 配 concurrencyPolicy: Forbid——这一条就天然解决了防重入问题（上一个 Job 没结束就不启动新的）。③ 处理 SIGTERM——K8s 缩容或滚动更新时会先发 SIGTERM 再等宽限期（默认 30 秒）后 SIGKILL；捕获它并只置标志位、跑完当前批次再退出，能避免数据处理到一半被强杀。④ 日志直接写 stdout/stderr——不要写文件（容器销毁就没了），让日志收集器（Fluentd/Loki）去采。⑤ 资源限制和超时——给 Job 设 activeDeadlineSeconds 防止卡死的任务永远占资源，同时确保内存 limit 够用（批处理容易 OOM，用 iterator(chunk_size=) 控制）。另外注意每个 Job 是独立 Pod，所以任何依赖本地文件锁的防重入方案在这里都失效，必须用分布式锁或 Forbid 策略。

八、加强记忆

管理命令 = Django 环境自动就绪的脚本入口。三件事定形态：<app>/management/commands/<名>.py（两层目录都必须有 __init__.py，缺了就静默找不到——新手最常卡住的地方） + class Command(BaseCommand) + handle(self, *args, **options)。相比独立脚本，它自动做了：加载 settings、django.setup() 让 app registry 就绪（独立脚本里模型必须在 setup() 之后 import，否则 AppRegistryNotReady）、argparse 解析、系统检查、关闭连接，还提供了一批内置参数（--settings/-v/--traceback/--skip-checks）。三条硬规矩：① 参数用 add_arguments(parser)（底层就是 argparse，注意 --dry-run 在 options 里叫 dry_run，横线变下划线）；② 输出用 self.stdout.write() 而不是 print——因为它可被 call_command(stdout=...) 重定向所以可测试、支持 self.style.SUCCESS/WARNING/ERROR 且非 TTY 时自动关色、能配合 verbosity 做分级；③ 出错 raise CommandError（简洁的红色输出 + 退出码 1，cron 和 CI 靠退出码判断成败），别静默 return（退出码 0 会掩盖失败）也别裸抛异常。生产四要素：幂等（cron 会重复触发，用 update_or_create/update_conflicts，别裸 create() 或 F()+1）、防重入（文件锁 / cache.add 原子锁 / select_for_update(nowait=True) / K8s 的 concurrencyPolicy: Forbid）、断点续传（超过 10 分钟的命令都该能续跑）、监控（退出码告警 + 死信心跳）。事务用分批各自 atomic 而不是整个命令一个大事务；注意 output_transaction 不是「包事务」，它只让命令输出的 SQL 文本被 BEGIN/COMMIT 包住。最重要的设计原则是「命令只做入口，业务逻辑放 services 层」——这样同一份逻辑能被命令、Celery 任务、admin action、API 复用，也更好测试（handle() 超过 50 行就该抽了）。crontab 有四个坑：虚拟环境的绝对路径 + cd 到项目目录 + 重定向日志带 2>&1 + 显式 --settings。测试写法是 call_command("cmd", 30, dry_run=True, stdout=StringIO())（参数名用下划线）。