代码之家 › 专栏 › 技术社区 › NeoVe

芹菜调度Django法

celery django python

NeoVe · 技术社区 · 6 年前

我有这个方法:

def getExchangeRates(): 
    """ Here we have the function that will retrieve the latest rates from fixer.io """
    rates = {}
    response = urlopen('http://data.fixer.io/api/latest?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']:
        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 index(request):
    rates = getExchangeRates()
    fixerio_rates = [Fixerio_rates(currency=currency, rate=rate)
                 for currency, rate in rates.items()]
    Fixerio_rates.objects.bulk_create(fixerio_rates)
    return render(request, 'index.html')

我想把它安排在每天早上9点,周末除外。

我还没有找到关于如何根据这样一个特定的日期时间安排这个方法的全面教程,而且,我不知道我是否可以安排这个方法,或者在我的系统中创建另一个方法 tasks 继承此文件并在任何特定日期运行的文件。

我确实有 celery.py tasks.py

或者,也许芹菜不是解决这种情况的方法?

有什么想法吗?

3 回复 | 直到 6 年前

Thiago F. Pappacena 6 年前

有一些django包允许您使用django管理界面管理“类似cron”的作业。我过去用过django计时表和django计时器( https://github.com/chrisspen/django-chroniker ).还有django cron( https://django-cron.readthedocs.io/en/latest/installation.html

它们都有类似的方法:在crontab上创建一个条目,运行类似 python manage.py runcrons 每一分钟,每一天 settings.py

查看Chroniker或Django cron的文档,了解有关如何设置它的更多信息。

Artem Rys 5 年前

此外,您还可以使用 Celery Beat

Francisco Puga 5 年前

任务可以通过 Celery Beat .

芹菜酱必须作为另一道工序推出。这 beat 进程将把计划的任务踢到 celery worker process 这将像其他芹菜异步任务一样启动任务。要了解这两个过程通常是个好主意 supervisord 在生产和 honcho 正在开发中。

计划任务可以在代码中定义,也可以存储在数据库中,并通过django admin和扩展名进行处理 django-celery-beat

要通过代码添加它,最简单的方法是在 tasks.py

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):

    # Executes every Monday morning at 9 a.m.
    sender.add_periodic_task(
        crontab(hour=9, minute=0, day_of_week=1),
        test.s('Happy Mondays!'),
    )
    sender.add_periodic_task(
        crontab(hour=9, minute=0, day_of_week=2),
        test.s('Happy Tuesday!'),
    )
    sender.add_periodic_task(
        crontab(hour=9, minute=0, day_of_week=3),
        test.s('Happy Wednesday!'),
    )
    sender.add_periodic_task(
        crontab(hour=9, minute=0, day_of_week=4),
        test.s('Happy Thursday!'),
    )
    sender.add_periodic_task(
        crontab(hour=9, minute=0, day_of_week=1),
        test.s('Happy Friday!'),
    )

@app.task
def test(arg):
    print(arg)