python之如何将基于类的任务传递给 CELERY_BEAT_SCHEDULE

如 docs 中所示基于类的任务是表达复杂逻辑的一种公平方式。

但是，文档没有指定如何将新创建的 Shiny 的基于类的任务添加到您的 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/

请您参考如下方法:

我也花了一段时间才找到这个问题的答案，因为这个问题在谷歌搜索结果中排名靠前，我想我会把它放在这里给那些正在努力寻找答案的人:

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

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 ...