セロリクラスベースタスクの登録

Question

Python 3.x、Celery 4.x ...

私はクラスベースのタスクを持っています。

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 

ので、セロリのデコレータは、タスクを登録するために動作しますが、クラスベースのタスクが登録されていません。

クラスベースのタスクを登録するにはどうすればよいですか？

アップデート1：

私は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：

私はラッパーを経由して同期（run）私のタスクを実行することができています。しかし、私はタスクを非同期、つまりdelayで実行することはできません。

app1/tasks.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. 

Answer 1

あなたは完全な説明hereを見つけることができますが、私にとって、

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