如 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()
我查过但没有帮助我/解决问题的在线资源:
我也花了一段时间才找到这个问题的答案,因为这个问题在谷歌搜索结果中排名靠前,我想我会把它放在这里给那些正在努力寻找答案的人:
您可以像添加普通任务一样添加它,但使用类名。
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 ...
我是一名优秀的程序员,十分优秀!