如何在芹菜link_error回调中获得“完全”异步结果
如何在芹菜link_error回调中获得“完全”异步结果
提问于 2015-08-13 13:16:05
我在Django 1.6.11和RabbitMQ 3.5.4中运行了芹菜3.1.18，并试图在失败状态(CELERY_ALWAYS_EAGER=True)中测试异步任务。但是，我无法在错误回调中获得正确的“结果”。芹菜医生中的示例显示：
@app.task(bind=True)
def error_handler(self, uuid):
result = self.app.AsyncResult(uuid)
print('Task {0} raised exception: {1!r}\n{2!r}'.format(
uuid, result.result, result.traceback))当我这样做时，我的结果仍然是“待定”、result.result = ''和result.traceback=''。但是，我的.apply_async调用返回的实际结果具有正确的"FAILURE“状态和回溯。
我的代码(基本上是一个Django Rest框架RESTful端点，它解析一个.tar.gz文件，然后在文件解析完成时向用户发送一个通知)：
views.py：
from producer_main.celery import app as celery_app
@celery_app.task()
def _upload_error_simple(uuid):
print uuid
result = celery_app.AsyncResult(uuid)
print result.backend
print result.state
print result.result
print result.traceback
msg = 'Task {0} raised exception: {1!r}\n{2!r}'.format(uuid,
result.result,
result.traceback)
class UploadNewFile(APIView):
def post(self, request, repository_id, format=None):
try:
uploaded_file = self.data['files'][self.data['files'].keys()[0]]
self.path = default_storage.save('{0}/{1}'.format(settings.MEDIA_ROOT,
uploaded_file.name),
uploaded_file)
print type(import_file)
self.async_result = import_file.apply_async((self.path, request.user),
link_error=_upload_error_simple.s())
print 'results from self.async_result:'
print self.async_result.id
print self.async_result.backend
print self.async_result.state
print self.async_result.result
print self.async_result.traceback
return Response()
except (PermissionDenied, InvalidArgument, NotFound, KeyError) as ex:
gutils.handle_exceptions(ex)tasks.py：
from producer_main.celery import app
from utilities.general import upload_class
@app.task
def import_file(path, user):
"""Asynchronously import a course."""
upload_class(path, user)celery.py：
"""
As described in
http://celery.readthedocs.org/en/latest/django/first-steps-with-django.html
"""
from __future__ import absolute_import
import os
import logging
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'producer_main.settings')
from django.conf import settings
log = logging.getLogger(__name__)
app = Celery('producer') # pylint: disable=invalid-name
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) # pragma: no cover
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))我的后端配置如下：
CELERY_ALWAYS_EAGER = True
CELERY_EAGER_PROPAGATES_EXCEPTIONS = False
BROKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'redis://localhost'
CELERY_RESULT_PERSISTENT = True
CELERY_IGNORE_RESULT = False当我为link_error状态运行我的单元测试时，我得到：
Creating test database for alias 'default'...
<class 'celery.local.PromiseProxy'>
130ccf13-c2a0-4bde-8d49-e17eeb1b0115
<celery.backends.redis.RedisBackend object at 0x10aa2e110>
PENDING
None
None
results from self.async_result:
130ccf13-c2a0-4bde-8d49-e17eeb1b0115
None
FAILURE
Non .zip / .tar.gz file passed in.
Traceback (most recent call last):因此，任务结果在我的_upload_error_simple()方法中不可用，但它们可以从self.async_result返回的变量中获得.
回答 3
Stack Overflow用户
回答已采纳
发布于 2015-08-14 16:54:58
我无法让link和link_error回调工作，所以我最终不得不使用医生们和这个问题中描述的on_failure和on_success任务方法。然后，我的tasks.py看起来像：
class ErrorHandlingTask(Task):
abstract = True
def on_failure(self, exc, task_id, targs, tkwargs, einfo):
msg = 'Import of {0} raised exception: {1!r}'.format(targs[0].split('/')[-1],
str(exc))
def on_success(self, retval, task_id, targs, tkwargs):
msg = "Upload successful. You may now view your course."
@app.task(base=ErrorHandlingTask)
def import_file(path, user):
"""Asynchronously import a course."""
upload_class(path, user)Stack Overflow用户
发布于 2015-08-13 16:35:39
您似乎将_upload_error()作为类的绑定方法--这可能不是您想要的。试着让它成为一项持久的任务：
@celery_app.task(bind=True)
def _upload_error(self, uuid):
result = celery_app.AsyncResult(uuid)
msg = 'Task {0} raised exception: {1!r}\n{2!r}'.format(uuid,
result.result,
result.traceback)
class Whatever(object):
....
self.async_result = import_file.apply_async((self.path, request.user),
link=self._upload_success.s(
"Upload finished."),
link_error=_upload_error.s())实际上，没有必要使用self参数，因为它不被使用，所以您可以这样做：
@celery_app.task()
def _upload_error(uuid):
result = celery_app.AsyncResult(uuid)
msg = 'Task {0} raised exception: {1!r}\n{2!r}'.format(uuid,
result.result,
result.traceback)注意没有bind=True和self
Stack Overflow用户
发布于 2020-06-17 05:22:37
小心UUID实例！
如果您尝试获得id而不是string类型而是UUID类型的任务的状态，则只能获得PENDING状态。
from uuid import UUID
from celery.result import AsyncResult
task_id = UUID('d4337c01-4402-48e9-9e9c-6e9919d5e282')
print(AsyncResult(task_id).state)
# PENDING
print(AsyncResult(str(task_id)).state)
# SUCCESS页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/31989366
复制相关文章
相似问题
腾讯云开发者
