Python Celeryタイプインポートアプリケーションエラーの登録解除されたタスクを受け取りました

Question

メインのPythonアプリケーションからタスクを実行するためにセロリアプリケーションをインポートできません。私はmyprogram.pyファイルからセロリのタスクを実行できるようにしたい。次のように

マイcelery_app.pyファイルは次のとおりです。

import celery 

app = celery.Celery('MyApp', broker='redis://localhost:6379/0') 

app.conf.broker_url = 'redis://localhost:6379/0' 
app.conf.result_backend = 'redis://localhost:6379/0' 
app.autodiscover_tasks() 

@app.task(ignore_result=True)  
def task_to_run(): 
    print("Task Running") 

# The following call runs a worker in celery 
task_to_run.delay() 

if __name__ == '__main__': 
    app.start() 

アプリケーション構造

• projectfolder /コア/ celery_app.pyの＃セロリアプリ
• projectfolder /コア/ myprogram.py＃マイPythonアプリケーション
• プロジェクトフォルダ/コア/その他のpythonファイル...

ファイルmyprogram.pyには、次のものが含ま

from .celery_app import task_to_run 
task_to_run.delay() 

エラー：

Answer 1

興味深い

Received unregistered task of type 'projectfolder.core.celery_app.task_to_run'. 
The message has been ignored and discarded. 

Did you remember to import the module containing this task? 
Or maybe you're using relative imports? 

    strategy = strategies[type_] 
KeyError: 'projectfolder.core.celery_app.task_to_run' 

おかげで、私はそれが4.1

に新しいだと思う、およそ autodiscover_tasksを知りませんでした

documentationにあるように、この関数は検索するパッケージのリストを取ります。

app.autodiscover_tasks(['core.celery_app'])

するか、tasks.pyと呼ばれる別々のファイルにタスクを抽出する方がよいかもしれませんし、それだけで次のようになります：あなたはでそれを呼び出すこともできます app.autodiscover_tasks(['core'])。 Celeryインスタンスを作成するときに

また、あなたはinculdeパラメータを使用することができます。

app = celery.Celery('MyApp', broker='redis://localhost:6379/0', include=['core.celery_app'])かを、あなたのタスクがどこにいても。

幸運