Django commands自定制

什么是Django Commands

Django 對於命令的添加有一套規范，你可以為每個app 指定命令。通俗一點講，比如在使用manage.py文件執行命令的時候，可以自定制自己的命令，來實現命令的擴充。

 

commands的創建

1、在app內創建一個management的python目錄
2、在management目錄里面創建commands的python文件夾
3、在commands文件夾下創建任意py文件

此時py文件名就是你的自定制命令，可以使用下面方式執行

python manage.py 命令名

 

撰寫command文件要求 

首先對於文件名沒什么要求，內部需要定義一個Command類並繼承BaseCommand類或其子類。

文件結構

其中help是command功能作用簡介，handle函數是主處理程序，add_arguments函數是用來接收可選參數的

簡單測試

# -*- coding: utf-8 -*-
# __author__ = 'dandy'
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = 'test'

    def handle(self, *args, **options):
        print('test')

帶參數的測試

# -*- coding: utf-8 -*-
# __author__ = 'dandy'
from django.core.management.base import BaseCommand

class Command(BaseCommand):

    def add_arguments(self, parser):
        parser.add_argument('aaa', nargs='+', type=int)
        parser.add_argument('--delete',
                            action='store_true',
                            dest='delete',
                            default=False,
                            help='Delete poll instead of closing it')

    def handle(self, *args, **options):
        print('test')
        print(args, options)

options里面直接取參數就可以了。

# -*- coding: utf-8 -*-
# __author__ = 'dandy'
from django.core.management.base import BaseCommand
from django.conf import settings
import requests
import os
import threading

command_path = os.path.join(settings.BASE_DIR, 'api', 'management', 'commands')
file = os.path.join(command_path, 'api_urls')


class Command(BaseCommand):
    help = 'test cost time for each api when getting data .We have set a middleware and create a file for log.' \
           'here just send request by thread pool'

    def handle(self, *args, **options):
        url_list = []
        try:
            if not os.path.exists(file):
                raise FileNotFoundError('ERROR！！！！ no file named api_urls in %s' % file)
            with open (file, 'rb') as obj:
                urls = obj.readlines()
                if not len(urls):
                    raise Exception('ERROR！！！ api_urls is a empty file !!')
                for url in urls:
                    url = url.strip()
                    if url:
                        t = threading.Thread(target=self.get_url, args=(url,))
                        # t.setDaemon(True)  # set True and you don't need to wait until main thread finished
                        t.start()
            print('send all request successfully !')
        except Exception as e:
            print(e)

    def get_url(self, url):
        requests.get(url)

http://www.baidu.com
http://www.qq.com