注册基于Celery Class的任务 [英] Register Celery Class-based Task

问题描述

Python 3.x，Celery 4.x ...

Python 3.x, Celery 4.x...

我有一个基于类的任务。

I have a class-based task.

myproj / celery.py

from celery import Celery

# django settings stuff...

app = Celery('myproj')
app.autodiscover_tasks()

app1 / tasks.py

import celery

class EmailTask(celery.Task):
    def run(self, *args, **kwargs):
        self.do_something()

如果我这样做：

$ celery worker -A myproj -l info
[tasks]
  . app2.tasks.debug_task
  . app2.tasks.test

因此，芹菜装饰器可以注册任务，但基于类的任务未注册。

So, the celery decorators work to register tasks, but the class-based task is not registered.

更新1：

如果我将以下行添加到 app1 / tasks.py

If I add the following lines to app1/tasks.py

from myproj.celery import app
email_task = app.tasks[EmailTask.name]

。

$ celery worker -A myproj -l info
  File "myproj/app1/tasks.py", line 405, in <module>
    email_task = app.tasks[EmailTask.name]
  File "/usr/local/lib/python3.5/site-packages/celery/app/registry.py", line 19, in __missing__
    raise self.NotRegistered(key)
celery.exceptions.NotRegistered

更新2：

我能够通过包装程序同步执行任务（运行） 。但是，我无法异步运行任务，即通过 delay 。

I am able to execute my task synchronously (run) via a wrapper. However, I cannot run the task async, i.e., via delay.

app1 / task.py

@app.task
def email_task():
    """
    Wrapper to call class based task
    """
    task = EmailTask()
    # task.delay()  # Won't work!!!
    task.run()

。

$./manage.py shell
> from app1.tasks import EmailTask
> task1 = EmailTask()
> task1.run() # a-okay
> task2 = EmailTask()
> task2.delay() # nope
  <AsyncResult: 1c03bad9-169a-4a4e-a56f-7d83892e8bbc>

# And on the worker...
[2017-01-22 08:07:28,120: INFO/PoolWorker-1] Task app1.tasks.email_task[41e5bc7d-058a-400e-9f73-c853c0f60a2a] succeeded in 0.0701281649817247s: None
[2017-01-22 08:10:31,909: ERROR/MainProcess] Received unregistered task of type None.
The message has been ignored and discarded.

推荐答案

您可以找到完整的说明在这里，但对我来说，足以添加

You can find full description here, but for me it was enough to add

from myapp.celery import app
app.tasks.register(MyTaskTask())

这篇关于注册基于Celery Class的任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！