Celery Result错误"args必须是列表或元组". [英] Celery Result error "args must be a list or tuple"

问题描述

我正在运行Django网站，刚刚让Celery可以运行，但是我遇到了令人困惑的错误.这是代码的结构.

I am running a Django website and have just gotten Celery to run, but I am getting confusing errors. Here is how the code is structured.

在tests.py中:

In 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中:

in 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中:

in mainapp/settings.py:

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

必须按列表/元组的celery调试控制台日志:

the celery debug console log for must by list/tuple:

[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:任务args必须是列表或元组.这应该是一个简单的解决方案，但事实并非如此.当我将其设为这样的列表时:

the error I am getting is as above, AsyncResult(c.id).result: task args must be a list or tuple. This should be an easy solution but it is not. When I make it a list like so:

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

然后它改变了主意，并告诉我 AsyncResult(c.id).result:int()参数必须是字符串或数字，而不是列表"

It then changes it mind and tells me that AsyncResult(c.id).result: int() argument must be a string or a number, not 'list'

您可以想象，我对可能是什么问题感到非常困惑.

As you can imagine I am very confused as to what might be the problem.

tasks.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打印

function.app prints

推荐答案

您在任务中的代码中遇到错误，它在回溯中显示:

You are getting an error in your code inside the task, it shows in traceback:

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

您的代码似乎正确，但是从追溯中看，芹菜似乎已腌制了旧版本的任务.非常重要，只要您更改 task.py 中的任何内容，就可以重新启动celery(即使您更改了其他文件，但我也不认为如此).可能是您遇到问题的原因，它使我伤了几次.

Your code seems right, but from the traceback it looks like celery has an old version of task pickled. Very important that you restart celery whenever you change anything inside task.py (maybe even if you change other files, but I don't think so). It could be a cause of your problem, it bit me in a butt couple of times.

也没有理由单独从数据库中提取 part p = Part.objects.get(id = part.id)，因为您已经获得了当前零件实例当您遍历中的查询集以获取project.part_set.all():中的一部分时.这只是一个建议，您可能需要执行此步骤的更多代码.

Also there is no reason to pull the part from database individually p = Part.objects.get(id=part.id) since you already getting current part instance when you are iterating over a queryset in for part in project.part_set.all():. This one is just a suggestion, you might have more code that needs that step.

请注意，如果这是您项目中的任务，而不是某些可重用应用程序的一部分，则只需使用常规的 @task 装饰器，celery会找到它，但请确保您进行了配置 Celery 应用程序对，这是我的另一篇文章，可以指导您完成以下操作:

As a side note, if it is a task in your project and not a part of some reusable app, just use a regular @task decorator, celery will find it, but make sure you configure Celery app right, here is another one of my posts that can guide you through: Celery / Django Single Tasks being run multiple times

只要您正确配置了所有内容，就可以像以前一样使用它:

So as long as you have everything configured correctly, just use it as you did before:

@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,))

很明显，我也建议您通过不带芹菜 function(project.id)的方式直接调用任务来测试任务，但是我敢肯定您知道这一点.

Obviously I'd also recommend testing the task directly by calling it without celery function(project.id), but I'm sure you knew that.

这篇关于Celery Result错误"args必须是列表或元组".的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！