django如何自定義manage.py管理命令
每次在啟動Django服務之前,我們都會在終端運行python manage.py xxx的管理命令。其實我們還可以自定義管理命令,這對於執行獨立的腳本或任務非常有用,比如清除緩存、導出用戶郵件清單或發送郵件等等。
自定義的管理命令不僅可以通過manage.py運行,還可以通過Linux或Celery的crontab服務將其設成定時任務。本文主要講解如何自定義Django-admin命令,並提供一些演示案例。
自定義Django-admin命令一共分三步:創建文件夾佈局、編寫命令代碼和測試使用。
創建文件夾佈局
自定義的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 = "[email protected]" ADMINS = [("大江狗", "[email protected]"), ]
但是如果每天都要進入終端運行這個命令實在太麻煩瞭,我們完全可以使用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管理命令的資料請關註WalkonNet其它相關文章!
推薦閱讀:
- Django定時任務Django-crontab的使用詳解
- Django rest framework如何自定義用戶表
- django配置DJANGO_SETTINGS_MODULE的實現
- Django數據庫遷移報錯InconsistentMigrationHistory
- 快速上手基於Anaconda搭建Django環境的教程