小编Doc*_*ard的帖子

我对celery和django一般都是新手,所以请原谅我缺乏知识.我正在尝试运行测试来进行一些计算并等待测试完成,这样我就能确保完成正确的答案.

这是我有的:

在app/tests.py中

from tasks import *


c = calculate.apply_async(args=[1])

# wait until the task is done
while not calculate.AsyncResult(c.id).status == "SUCCESS":
    print c.state
    pass

在app/tasks.py中

from celery import shared_task

@shared_task
def calculate(proj_id):

    #some calculations followed by a save of the object

即使在芹菜日志中它表示任务已成功完成,状态也永远不会从挂起更改

[2014-06-10 17:55:11,417: INFO/MainProcess] Received task: app.tasks.calculate[1f11e7ab-0add-42df-beac-3d94c6868aac]
[2014-06-10 17:55:11,505: INFO/MainProcess] Task app.tasks.calculate[1f11e7ab-0add-42df-beac-3d94c6868aac] succeeded in 0.0864518239978s: None

我还在mainapp/settings.py中放了CELERY_IGNORE_RESULT = False,但这似乎没有做任何事情.

我正在运行一个Django网站并且刚刚让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) …