您好,登錄后才能下訂單哦!
這篇文章主要介紹了django如何自定義manage.py管理命令,具有一定借鑒價(jià)值,感興趣的朋友可以參考下,希望大家閱讀完這篇文章之后大有收獲,下面讓小編帶著大家一起了解一下。
每次在啟動(dòng)Django服務(wù)之前,我們都會(huì)在終端運(yùn)行python manage.py xxx的管理命令。其實(shí)我們還可以自定義管理命令,這對于執(zhí)行獨(dú)立的腳本或任務(wù)非常有用,比如清除緩存、導(dǎo)出用戶郵件清單或發(fā)送郵件等等。
自定義的管理命令不僅可以通過manage.py運(yùn)行,還可以通過Linux或Celery的crontab服務(wù)將其設(shè)成定時(shí)任務(wù)。本文主要講解如何自定義Django-admin命令,并提供一些演示案例。
自定義Django-admin命令一共分三步:創(chuàng)建文件夾布局、編寫命令代碼和測試使用。
自定義的Django-admin管理命令本質(zhì)上是一個(gè)python腳本文件,它的存放路徑必須遵循一定的規(guī)范,一般位于app/management/commands目錄。整個(gè)文件夾的布局如下所示:
app01/ __init__.py models.py management/ __init__.py commands/ __init__.py _private.py # 以下劃線開頭文件不能用作管理命令 my_commands.py # 這個(gè)就是自定義的管理命令腳本,文件名即為命令名 tests.py views.py
注意:
management和commands每個(gè)目錄下都必須有個(gè)__init__.py空文件,表明這是一個(gè)python包。另外以下劃線開頭的文件名不能用作管理命令腳本。
management/commands目錄可以位于任何一個(gè)app的目錄下,Django都能找到它。
一般建議每個(gè)python腳本文件對應(yīng)一條管理命令。
每一個(gè)自定義的管理命令本質(zhì)是一個(gè)Command類, 它繼承了Django的Basecommand或其子類, 主要通過重寫handle()方法實(shí)現(xiàn)自己的業(yè)務(wù)邏輯代碼,而add_arguments()則用于幫助處理命令行的參數(shù),如果運(yùn)行命令時(shí)不需要額外參數(shù),可以不寫這個(gè)方法。
from django.core.management.base import BaseCommand class Command(BaseCommand): # 幫助文本, 一般備注命令的用途及如何使用。 help = 'Some help texts' # 處理命令行參數(shù),可選 def add_arguments(self, parser): pass # 核心業(yè)務(wù)邏輯 def handle(self, *args, **options): pass
我們現(xiàn)在來看一個(gè)最簡單的例子,希望定義一個(gè)名為hello_world的命令。這樣當(dāng)我們運(yùn)行python manage.py hello_world命令時(shí),控制臺會(huì)打印出Hello World!字樣。在app/management/commands目錄下新建hello_world.py, 添加如下代碼:
from django.core.management.base import BaseCommand class Command(BaseCommand): # 幫助文本, 一般備注命令的用途及如何使用。 help = "Print Hello World!" # 核心業(yè)務(wù)邏輯 def handle(self, *args, **options): self.stdout.write('Hello World!')
注意:當(dāng)你使用管理命令并希望在控制臺輸出指定信息時(shí),你應(yīng)該使用self.stdout和self.stderr方法,而不能直接使用python的print方法。另外,你不需要在消息的末尾加上換行符,它將被自動(dòng)添加。
此時(shí)當(dāng)你進(jìn)入項(xiàng)目文件夾運(yùn)行python manage.py hello_world命令時(shí),你將得到如下輸出結(jié)果:
現(xiàn)在我們來增加點(diǎn)難度,來通過命令行給hello_world命令傳遞一個(gè)name參數(shù),以實(shí)現(xiàn)運(yùn)行python manage.py helloworld John命令時(shí) 打印出Hello World! John。
現(xiàn)在修改我們的hello_world.py, 添加add_arguments方法,該方法的作用是給自定義的handle方法添加1個(gè)或多個(gè)參數(shù)。
from django.core.management.base import BaseCommand class Command(BaseCommand): # 幫助文本, 一般備注命令的用途及如何使用。 help = "Print Hello World!" # 給命令添加一個(gè)名為name的參數(shù) def add_arguments(self, parser): parser.add_argument('name') # 核心業(yè)務(wù)邏輯,通過options字典接收name參數(shù)值,拼接字符串后輸出 def handle(self, *args, **options): msg = 'Hello World ! '+ options['name'] self.stdout.write(msg)
此時(shí)當(dāng)你再次運(yùn)行python manage.py hello_world John命令時(shí),你將得到如下輸出結(jié)果:
如果你直接運(yùn)行命令而不攜帶參數(shù),將會(huì)報(bào)錯(cuò),如下所示:
前面的案例過于簡單,我們現(xiàn)在來看兩個(gè)自定義管理命令的實(shí)際應(yīng)用案例。
無論你使用常規(guī)方式還是Docker在生產(chǎn)環(huán)境中部署Django項(xiàng)目,你需要確保數(shù)據(jù)庫連接已就緒后才進(jìn)行數(shù)據(jù)庫遷移(migrate)的命令(Docker-compose的depends選項(xiàng)并不能確保這點(diǎn)),否則Django應(yīng)用程序會(huì)出現(xiàn)報(bào)錯(cuò)。
這時(shí)你可以自定義一個(gè)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!'))
定義好這個(gè)命令后每次在運(yùn)行python manage.py migrate命令前先運(yùn)行python manage.py wait_for_db即可。
如果你是網(wǎng)站管理員,你肯定希望知道每天有多少新用戶已注冊,這時(shí)你可以自定義一條mail_admin的管理命令,將每天新注冊用戶數(shù)量以郵件形式發(fā)給自己,如下所示:
# 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): # 獲取過去一天注冊用戶數(shù)量 user_count =User.objects.filter(date_joined__range=(yesterday, today)).count() # 當(dāng)注冊用戶數(shù)量多余1個(gè),才發(fā)送郵件給管理員 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.")
如果你在終端運(yùn)行python manage.py mail_admin命令,你將得到如下輸出結(jié)果:
注意:真正發(fā)送郵件成功需要設(shè)置Email后臺及管理員,測試環(huán)境下可以使用如下簡單配置:
EMAIL_BACKEND = "django.core.mail.backends.console.EmailBackend" DEFAULT_FROM_EMAIL = "noreply@example.com" ADMINS = [("大江狗", "yunbo.shi@example.com"), ]
但是如果每天都要進(jìn)入終端運(yùn)行這個(gè)命令實(shí)在太麻煩了,我們完全可以使用Linux的crontab服務(wù)或Celery-Beat將其設(shè)成周期性定時(shí)任務(wù)task,這時(shí)只需要調(diào)用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", )
感謝你能夠認(rèn)真閱讀完這篇文章,希望小編分享的“django如何自定義manage.py管理命令”這篇文章對大家有幫助,同時(shí)也希望大家多多支持億速云,關(guān)注億速云行業(yè)資訊頻道,更多相關(guān)知識等著你來學(xué)習(xí)!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。