运行 django_cron 上的特定 views.py 方法

Question

我的 views.py 文件中有这个方法：

def getHistoricRates():
    """ Here we have the function that will retrieve the historical rates from fixer.io, since last month """
    rates = {}
    response = urlopen('http://data.fixer.io/api/2018-12-31?access_key=c2f5070ad78b0748111281f6475c0bdd')
    data = response.read()
    rdata = json.loads(data.decode(), parse_float=float) 
    rates_from_rdata = rdata.get('rates', {})
    for rate_symbol in ['USD', 'GBP', 'HKD', 'AUD', 'JPY', 'SEK', 'NOK', 'EUR']:
        try:
            rates[rate_symbol] = rates_from_rdata[rate_symbol]
        except KeyError:
            logging.warning('rate for {} not found in rdata'.format(rate_symbol)) 
            pass

    return rates

@require_http_methods(['GET', 'POST'])
def historical(request):
    date_str = "2018-12-31"
    if datetime.datetime.strptime(date_str,"%Y-%m-%d").weekday()<5:
        rates = getHistoricRates()
        fixerio_rates = [Fixerio_rates(currency=currency, rate=rate)
                     for currency, rate in rates.items()]
        Fixerio_rates.objects.bulk_create(fixerio_rates) 
        return render(request, 'historical.html')

我想运行 historical 每天早上 9 点，周末除外。

现在，我找不到任何关于如何运行现有方法或如何从 cron.py 文件中调用它的示例。

我确实为 django_cron 配置了所有内容，但我无法弄清楚如何在特定时间从我的 cron 文件 "use" 到运行它。

这是我目前的 cron.py 文件：

from django_cron import CronJobBase, Schedule
from .views import historical

class MyCronJob(CronJobBase):
    RUN_AT_TIMES = ['9:00']

    schedule = Schedule(run_at_times=RUN_AT_TIMES)
    code = 'fixerio.my_cron_job'    # a unique code

    def do(self):
        pass    # do your thing here

名称 fixerio 是我的应用程序的名称。

对此有什么想法吗？

Answer 1

由于您希望从 historical() 视图和 cron 作业中调用 getHistoricRates() 和 bulk_create() 逻辑，因此最好先重构那个通用的代码从视图到一个单独的模块 - 例如进入 helpers.py 与 views.py 和 cron.py.

一起生活

helpers.py

from .models import Fixerio_rates

def create_rates():
    rates = getHistoricRates()
    fixerio_rates = [Fixerio_rates(currency=currency, rate=rate)
                     for currency, rate in rates.items()]
    Fixerio_rates.objects.bulk_create(fixerio_rates) 

def getHistoricRates():
    ...

然后您可以从您的 cron 作业中调用它：

cron.py

from .helpers import create_rates

class MyCronJob(CronJobBase):
    RUN_AT_TIMES = ['9:00']

    schedule = Schedule(run_at_times=RUN_AT_TIMES)
    code = 'fixerio.my_cron_job'    # a unique code

    def do(self):
        create_rates()

还有你的看法：

views.py

from .helpers import create_rates

@require_http_methods(['GET', 'POST'])
def historical(request):
    date_str = "2018-12-31"
    if datetime.datetime.strptime(date_str,"%Y-%m-%d").weekday()<5:
        create_rates()
        return render(request, 'historical.html')

运行 django_cron 上的特定 views.py 方法

Running specific views.py method on django_cron

python

django

django-cron