使用 Celery 调度 Django 方法

Question

我有这个方法：

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 文件。

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

有什么想法吗？

Answer 1

有一些 django 包可以让您使用 django 管理界面管理 "cron-like" 作业。我过去使用过 django-chronograph 和 django-chroniker (https://github.com/chrisspen/django-chroniker). There is also django-cron (https://django-cron.readthedocs.io/en/latest/installation.html)，但我从未使用过它。

他们都有类似的方法：你在你的 crontab 上创建一个条目，每分钟运行 python manage.py runcrons，然后在你的 settings.py 上添加包以在管理员上显示它。

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

Answer 2

此外，您可以使用 Celery Beat 来安排您需要的任务。

Answer 3

可以使用 Celery Beat 安排任务。

Celery Beat 必须作为另一个进程启动。此 beat 进程会将计划任务踢到开发中的 celery worker process that will launch the tasks like any other celery asynchronous task. To orquestrate these two process usually is a good idea use something like supervisord in production and honcho。

定时任务可以在代码中定义，也可以存储在数据库中，通过扩展名为django-celery-beat

的django-admin处理

要通过代码添加它，最简单的方法是在 tasks.py 文件中创建另一个方法。满足您每天上午 9 点的要求，周末除外 "it could look like this"

@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)

使用 Celery 调度 Django 方法

Scheduling Django method with Celery

python

django

celery