自定义management命令

前面我们曾多次使用过形如python manage.py xxx的命令，例如runserver、migrate等，这些都是Django框架内置的管理命令。实际上，Django也允许我们自定义管理命令，用于执行自定义系统维护、定时任务、批处理等操作。这篇笔记我们介绍自定义管理命令的用法。

创建自定义命令

自定义命令必须放在APP的management/commands/包下，假设我们要创建一个名为my_command的命令，它就应按约定如下放置。

app
├── __init__.py
├── management
│   └── __init__.py
│   └── commands/
│       ├── __init__.py
│       └── my_command.py
└── ...

注意：my_command.py必须位于合法的Python包中，这意味着其同级和上层都要有__init__.py，且该APP必须在INSTALLED_APPS中注册。

一个自定义命令的例子代码如下。

from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = '一个简单的示例命令'

    def handle(self, *args, **options):
        self.stdout.write('Hello, Django!')

自定义命令需要以类的形式定义，并继承BaseCommand，其中类属性help是帮助信息，用户使用命令时可用--help参数打印帮助信息，handle()方法具体定义了自定义命令的执行逻辑，我们这里只是打印Hello, Django!信息。

此时我们可以使用命令行测试。

python manage.py my_command

信息打印样式

Django管理命令在控制台上打印消息时应该使用self.stdout.write()方法而非print()，前者是Django提供的用于management命令的标准输出接口，它可以集成到Django的单元测试中，而且对颜色有直接支持。

def handle(self, *args, **options):
    # 添加各种样式
    self.stdout.write(self.style.SUCCESS('操作成功'))
    self.stdout.write(self.style.WARNING('这是警告'))
    self.stdout.write(self.style.ERROR('出错了'))

    # 也可以直接写到stderr
    self.stderr.write('错误输出')

参数定义与解析

Django底层使用Python标准库的argparse来解析参数，我们可以通过覆写BaseCommand的add_arguments()方法来定义参数，方法内的parse就来自argparse，我们可以按照argparse的使用说明添加位置参数和可选参数。

位置参数

from django.core.management.base import BaseCommand


class Command(BaseCommand):
    help = '删除指定ID的用户'

    def add_arguments(self, parser):
        # 接收一个或多个 user_id
        parser.add_argument(
            'user_ids',        # 参数名
            nargs='+',         # 至少一个
            type=int,          # 自动转为 int
            help='要删除的用户ID列表',
        )

    def handle(self, *args, **options):
        user_ids = options['user_ids']
        for uid in user_ids:
            self.stdout.write(f'删除用户: {uid}')

对应的测试命令如下。

python manage.py delete_users 1 2 3

nargs的常用取值如下表。

可选参数

from django.core.management.base import BaseCommand


class Command(BaseCommand):
    help = "发送报告邮件"

    def add_arguments(self, parser):
        # --count：整数类型，有默认值
        parser.add_argument(
            "--count",
            type=int,
            default=10,
            help="发送的报告数量（默认 10）",
        )

        # --dry-run：布尔开关，它不需要具体指定参数，action="store_true"表示--dry-run出现即为True
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="演习模式，不实际发送",
        )

        # --email：接收多个值
        parser.add_argument(
            "--email",
            nargs="+",
            type=str,
            help="收件人邮箱，可传多个",
        )

    def handle(self, *args, **options):
        count = options["count"]
        dry_run = options["dry_run"]
        emails = options["email"] or []

        self.stdout.write(f"发送 {count} 份报告")
        self.stdout.write(f"收件人: {emails}")

        if dry_run:
            self.stdout.write(self.style.WARNING("演习模式，跳过实际发送"))
        else:
            self.stdout.write(self.style.SUCCESS("发送成功！"))

对应的测试命令如下。

python manage.py send_report --count 5 --email a@x.com b@x.com --dry-run

代码调用命令

Django中也支持我们直接通过代码调用封装的自定义management命令，不过它不是启动子进程，而是直接在同一个进程内调用，具体写法如下。

from django.core.management import call_command

call_command(
    "send_report",
    count=5,
    dry_run=True,
    email=["a@x.com", "b@x.com"],
)

call_command函数中，第1个参数是自定义命令名，后面的关键字参数都是命令的命名参数。