5

Django Web サイトを実行していて、Celery を実行したばかりですが、紛らわしいエラーが発生します。コードの構造は次のとおりです。

tests.py で:

from tasks import *
from celery.result import AsyncResult

project = Project.objects.create()
# initalize various sub-objects of the project

c = function.delay(project.id)
r = AsyncResult(c.id).ready()
f = AsyncResult(c.id).failed()
# wait until the task is done  
while not r and not f:
    r = AsyncResult(c.id).ready()
    f = AsyncResult(c.id).failed()

self.assertEqual() #will fail because task fails

tasks.py で: 

from __future__ import absolute_import
from celery import shared_task

@shared_task
def function(project_id)
    #a bunch of calculations followed by a save of the project
    project = Project.objects.get(project=project_id)

    for part in project.part_set.all():
        partFunction(part.id)
        p = Part.objects.get(id=part.id)
        # add to various variables in project from variables in p
    project.save()

mainapp/settings.py で:

BROKER_URL = "amqp://ipaddress"
CELERY_RESULT_BACKEND='amqp'
CELERY_ACCEPT_CONTENT = ['json','pickle','msgpack','yaml']
CELERY_IGNORE_RESULT = False

リスト/タプルによるマストのセロリデバッグコンソールログ：

[INFO/MainProcess] Received task: myapp.tasks.function[id]
[ERROR/MainProcess] Task myapp.tasks.function[id]
    raised unexpected: ValueError('task args must be a list or tuple',)
Traceback:
   File "/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
       R = retval = fun(*args, **kwargs)
   File "/python2.7/site-packages/celery/app/trace.py", line 437, in __protected_call__
       return self.run(*args, **kwargs)
   File "/myapp/tasks.py", line 28, in function
       p = Part.objects.get(id=part.id)
   File "/python2.7/site-packages/celery/app/task.py", line 555, in apply_async
       **dict(self._get_exec_options(), **options)
   File "/python2.7/site-packages/celery/app/base.py", line 351, in send_task
       reply_to=reply_to or self.oid, **options
   File "celery/app/amqp.py", line 252, in publish_task
       raise ValueError('task args must be a list or tuple')
ValueError: task args must be a list or tuple

私が得ているエラーは上記のとおりですAsyncResult(c.id).result: task args must be a list or tuple。これは簡単な解決策のはずですが、そうではありません。リストにすると次のようになります。

inline = [project.id]
c = function.delay(inline)

それから気が変わって、私にこう言いますAsyncResult(c.id).result: int() argument must be a string or a number, not 'list'

ご想像のとおり、何が問題なのか、私は非常に混乱しています。

編集

タスク.py

@shared_task
def function(app):
    @app.task(name='myapp.function', bind=True)
    def function(project_id):

tests.py

c = function.s(project.id).delay()

function.app の出力

4

1 に答える 1

8

タスク内のコードでエラーが発生しています。トレースバックに表示されます。

File "/myapp/tasks.py", line 28, in function
   p = Part.objects.get(id=part.id)

あなたのコードは正しいようですが、トレースバックから、セロリには古いバージョンのタスク ピクルスがあるようです。内部で何かを変更するたびにセロリを再起動することが非常に重要task.pyです（おそらく他のファイルを変更した場合でも、そうは思いません）。それはあなたの問題の原因かもしれません、それは私を数回お尻に噛みました.

また、 でクエリセットを反復処理しているときに現在のパーツ インスタンスを取得しているためpart、データベースから を個別にプルする必要はありません。これは単なる提案です。その手順が必要なコードが他にもあるかもしれません。p = Part.objects.get(id=part.id)for part in project.part_set.all():

補足として、それがプロジェクト内のタスクであり、再利用可能なアプリの一部ではない場合、通常のデコレーターを使用するだけで、セロリはそれを見つけますが、アプリを正しく@task構成するようにしてください。Celeryガイド: Celery / Django Single Tasks are running multiple times

したがって、すべてが正しく構成されている限り、以前と同じように使用してください。

@task #or @shared_task
def function(project_id)
    #a bunch of calculations followed by a save of the project
    project = Project.objects.get(project=project_id)
    ....

次に、それを呼び出します。

result = function.delay(project.id)

また：

result = function.apply_async(args=(project.id,))

明らかに、 celery なしでタスクを呼び出してタスクを直接テストすることもお勧めしますfunction(project.id)が、あなたはそれを知っていると確信しています。

于 2014-06-18T21:35:25.910 に答える