Django 每小时运行预定作业答案

【问题标题】：Django run scheduled jobs hourlyDjango 每小时运行预定作业
【发布时间】：2021-09-02 22:57:47
【问题描述】：

在我的项目中，我必须每小时在数据库中加载数据。我尝试了 celery 和 cron，但发现所有的东西都非常复杂并且总是会产生一些问题。我在 windows 中使用 pycharm。我是 django 新手，只需要一个简单的解决方案来每小时运行以下命令。

这会在数据库中加载数据。

“python manage.py fetchfiles”

management/commands/fetchfiles

from django.core.management.base import BaseCommand, CommandError
from dashboard.models import airdata as tdata
import requests
import json

class Command(BaseCommand):
    help = 'Fetches api data'

    """def add_arguments(self, parser):
        none"""


    def handle(self, *args, **options):

        #for a in range(0,1578,10):
        a = True
        offno = 0
        lst2=[]
        dict1={}
        while a == True:
            api_key = "579b464db66ec23bdd000001cdd3946e44ce4aad7209ff7b23ac571b"
            url = "https://api.data.gov.in/resource/3b01bcb8-0b14-4abf-b6f2-c1bfd384ba69?api-key={}&format=json&offset={}&limit=10".format(api_key,offno)
            response = requests.get(url)
            data = response.text
            a1=json.loads(data)
            for ele in a1['records']:
                if ele['pollutant_min'] == 'NA':
                    dict1['pol_min'] = 0
                else:
                    dict1['pol_min'] = int(ele['pollutant_min'])
                if ele['pollutant_max'] == 'NA':
                    dict1['pol_max'] = 0
                else:
                    dict1['pol_max'] = int(ele['pollutant_max'])
                if ele['pollutant_avg'] == 'NA':
                    dict1['pol_avg'] = 0
                else:
                    dict1['pol_avg'] = int(ele['pollutant_avg'])
                dict1['state'] = ele['state']
                dict1['city'] = ele['city']
                dict1['station'] = ele['station']
                dict1['time_vector'] = ele['last_update']
                lst2.append(dict1.copy())
            if a1["count"] < 10:
                a= False
            offno += 10
        airx = json.dumps(lst2, indent=1)
        tdata.objects.bulk_create([tdata(**vals) for vals in lst2])
        return airx

【问题讨论】：

旁注：我猜你不应该在互联网上发布 API 密钥

标签： python django cron

【解决方案1】：

你有两种方法

将其添加到 Unix 系统上的 crontab 或 Windows 上的计划任务中。
使用 Celery Beat

【讨论】：

【解决方案2】：

Celery 可能比你只需要像他每小时运行一次那样复杂。您已经发现了编写管理命令。只需将管理命令包装在 shell 脚本中，然后（在 Unix/Linux 上）让 cron 每小时运行一次。您需要确保 shell 脚本在成功时保持“安静”，但使任何失败都非常明显，因此它不会坐在那里失败而没有人注意到。

无法建议在 Windows 上做什么，但我认为它有任务调度。

【讨论】：