django如何自定义manage.py管理命令

创建文件夹布局

自定义的Django-admin管理命令本质上是一个python脚本文件，它的存放路径必须遵循一定的规范，一般位于app/management/commands目录。整个文件夹的布局如下所示：

 app01/    __init__.py    models.py    management/        __init__.py        commands/            __init__.py            _private.py # 以下划线开头文件不能用作管理命令            my_commands.py # 这个就是自定义的管理命令脚本，文件名即为命令名    tests.py    views.py

注意：

management和commands每个目录下都必须有个__init__.py空文件，表明这是一个python包。另外以下划线开头的文件名不能用作管理命令脚本。
management/commands目录可以位于任何一个app的目录下，Django都能找到它。
一般建议每个python脚本文件对应一条管理命令。

编写命令代码

每一个自定义的管理命令本质是一个Command类, 它继承了Django的Basecommand或其子类, 主要通过重写handle()方法实现自己的业务逻辑代码，而add_arguments()则用于帮助处理命令行的参数，如果运行命令时不需要额外参数，可以不写这个方法。

 from django.core.management.base import BaseCommand  class Command(BaseCommand):     # 帮助文本, 一般备注命令的用途及如何使用。     help = 'Some help texts'      # 处理命令行参数，可选     def add_arguments(self, parser):        pass      # 核心业务逻辑     def handle(self, *args, **options):         pass

我们现在来看一个最简单的例子，希望定义一个名为hello_world的命令。这样当我们运行python manage.py hello_world命令时，控制台会打印出Hello World!字样。在app/management/commands目录下新建hello_world.py, 添加如下代码：

 from django.core.management.base import BaseCommand  class Command(BaseCommand):    # 帮助文本, 一般备注命令的用途及如何使用。    help = "Print Hello World!"     # 核心业务逻辑    def handle(self, *args, **options):        self.stdout.write('Hello World!')

注意：当你使用管理命令并希望在控制台输出指定信息时，你应该使用self.stdout和self.stderr方法，而不能直接使用python的print方法。另外，你不需要在消息的末尾加上换行符，它将被自动添加。

此时当你进入项目文件夹运行python manage.py hello_world命令时，你将得到如下输出结果：

现在我们来增加点难度，来通过命令行给hello_world命令传递一个name参数，以实现运行python manage.py helloworld John命令时打印出Hello World! John。

现在修改我们的hello_world.py, 添加add_arguments方法，该方法的作用是给自定义的handle方法添加1个或多个参数。

 from django.core.management.base import BaseCommand  class Command(BaseCommand):    # 帮助文本, 一般备注命令的用途及如何使用。    help = "Print Hello World!"     # 给命令添加一个名为name的参数    def add_arguments(self, parser):        parser.add_argument('name')     # 核心业务逻辑，通过options字典接收name参数值，拼接字符串后输出    def handle(self, *args, **options):        msg = 'Hello World ! '+ options['name']        self.stdout.write(msg)

此时当你再次运行python manage.py hello_world John命令时，你将得到如下输出结果：

如果你直接运行命令而不携带参数，将会报错，如下所示：

实际应用场景

前面的案例过于简单，我们现在来看两个自定义管理命令的实际应用案例。

案例1：检查数据库连接是否已就绪

无论你使用常规方式还是Docker在生产环境中部署Django项目，你需要确保数据库连接已就绪后才进行数据库迁移(migrate)的命令(Docker-compose的depends选项并不能确保这点)，否则Django应用程序会出现报错。

这时你可以自定义一个wait_for_db的命令，如下所示：

 # app/management/commands/wait_for_db.py  import time from django.db import connections from django.db.utils import OperationalError from django.core.management import BaseCommand   class Command(BaseCommand):     help = 'Run data migrations until db is available.'      def handle(self, *args, **options):         self.stdout.write('Waiting for database...')         db_conn = None         while not db_conn:             try:                 # 尝试连接                 db_conn = connections['default']             except OperationalError:                 # 连接失败，就等待1秒钟                 self.stdout.write('Database unavailable, waiting 1 second...')                 time.sleep(1)          self.stdout.write(self.style.SUCCESS('Database available!'))

定义好这个命令后每次在运行python manage.py migrate命令前先运行python manage.py wait_for_db即可。

案例2：周期性发送邮件

如果你是网站管理员，你肯定希望知道每天有多少新用户已注册，这时你可以自定义一条mail_admin的管理命令，将每天新注册用户数量以邮件形式发给自己，如下所示：

 # app/management/commands/mail_admin.py  #-*- coding:utf-8 -*- from datetime import timedelta, time, datetime from django.core.mail import mail_admins from django.core.management import BaseCommand from django.utils import timezone from django.contrib.auth import get_user_model  User = get_user_model()  today = timezone.now() yesterday = today - timedelta(1)   class Command(BaseCommand):     help = "Send The Daily Count of New Users to Admins"      def handle(self, *args, **options):         # 获取过去一天注册用户数量         user_count =User.objects.filter(date_joined__range=(yesterday, today)).count()                  # 当注册用户数量多余1个，才发送邮件给管理员         if user_count >= 1:             message = "You have got {} user(s) in the past 24 hours".format(user_count)              subject = (                 f"New user count for {today.strftime('%Y-%m-%d')}: {user_count}"            )              mail_admins(subject=subject, message=message, html_message=None)              self.stdout.write("E-mail was sent.")         else:             self.stdout.write("No new users today.")

如果你在终端运行python manage.py mail_admin命令，你将得到如下输出结果：

注意：真正发送邮件成功需要设置Email后台及管理员，测试环境下可以使用如下简单配置：

 EMAIL_BACKEND = "django.core.mail.backends.console.EmailBackend" DEFAULT_FROM_EMAIL = "noreply@example.com" ADMINS = [("大江狗", "yunbo.shi@example.com"), ]

但是如果每天都要进入终端运行这个命令实在太麻烦了，我们完全可以使用Linux的crontab服务或Celery-Beat将其设成周期性定时任务task，这时只需要调用Django的call_command方法即可。

 # app/tasks.py, 可以任一app目录下新建task from celery import shared_task from django.core.management import call_command  @shared_task def mail_admin():     call_command("mail_admin", )

关于Django项目中如何使用Celery执行异步和周期性任务，请参加下篇Django进阶-异步和周期任务篇。

以上就是django如何自定义manage.py管理命令的详细内容，更多关于django 自定义manage.py管理命令的资料请关注其它相关文章！

最后更新于 2021-11-23 09:10:43 并被添加「」标签，已有位童鞋阅读过。

本站使用「署名 4.0 国际」创作共享协议，可自由转载、引用，但需署名作者且注明文章出处