如何将基于类的任务传递到 CELERY_BEAT_SCHEDULE

Question

如文档中所示，基于类的任务是表达复杂逻辑的公平方法。

但是，文档没有指定如何将闪亮的新创建的基于类的任务添加到您中

CELERY_BEAT_SCHEDULE

（使用django）

我尝试过的事情：

celery.py

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS, 'task_summary')
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    from payments.tasks.generic.payeer import PayeerPaymentChecker
    from payments.tasks.generic.ok_pay import OkPayPaymentChecker

    okpay_import = OkPayPaymentChecker()
    payeer_imprt = PayeerPaymentChecker()

    sender.add_periodic_task(60.0, okpay_import.s(),
                             name='OkPay import',
                             expires=30)

    sender.add_periodic_task(60.0, payeer_imprt.s(),
                             name='Payeer import',
                             expires=30)

-- 或 --

payments/task_summary.py

from tasks.generic.import import OkPayPaymentChecker, PayeerPaymentChecker
 run_okpay = OkPayPaymentChecker()
 run_payeer = PayeerPaymentChecker()

CELERY_BEAT_SCHEDULE = {
# yes, i did try referring to the class here
'check_okpay_payments': {

    'task': 'payments.tasks.task_summary.run_okpay',
    'schedule': timedelta(seconds=60),
},
'check_payeer_payments': {
    'task': 'payments.task_summary.run_payeer',
    'schedule': timedelta(seconds=60),
},
}

真的不知道该怎么办，恢复到类似：

payments/task_summary.py

from payments.tasks.generic.ok_pay import OkPayPaymentChecker
from payments.tasks.generic.payeer import PayeerPaymentChecker
from celery import shared_task


@shared_task
def run_payer():
    instance = PayeerPaymentChecker()
    return instance.run()


@shared_task
def run_okpay():
    instance = OkPayPaymentChecker()
    return instance.run()

我检查过的在线资源没有帮助我/解决问题：

• https://denibertovic.com/posts/celery-best-practices/

• https://blog.balthazar-rouberol.com/celery-best-practices

• http://shulhi.com/class-based-celery-task/

• http://jsatt.com/blog/class-based-celery-tasks/

Answer 1

我也花了一段时间才找到这个问题的答案，因为这个问题在谷歌搜索结果中的排名如此之高，我想我会把它放在这里，供那些正在努力寻找答案的人使用：

您可以像普通任务一样添加它，但使用类名称。

CELERY_BEAT_SCHEDULE = {
    'my_task_name': {
        'task': 'mymodule.tasks.MyTaskClass',
        'schedule': timedelta(seconds=60),
},

（假设您有

mymodule/tasks.py

和：

from celery import Task

class MyTaskClass(Task):

    def run(self, *args, **kwargs):
       ... stuff ...