gpt4 book ai didi

django - 如何在 Celery link_error 回调中获取 "full"异步结果

转载 作者:行者123 更新时间:2023-12-03 12:18:11 25 4
gpt4 key购买 nike

我将 Celery 3.1.18 与 Django 1.6.11 和 RabbitMQ 3.5.4 一起运行,并尝试在失败状态 (CELERY_ALWAYS_EAGER=True) 下测试我的异步任务。但是,我无法在错误回调中获得正确的“结果”。 Celery docs中的例子显示:

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

当我这样做时,我的结果仍然是“PENDING”, result.result = '' , 和 result.traceback='' .但是我的 .apply_async返回的实际结果调用具有正确的“失败”状态和回溯。

我的代码(基本上是解析 .tar.gz 文件的 Django Rest Framework RESTful 端点,然后在文件解析完成后将通知发送回用户):

View .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)

任务.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 获得返回变量...

最佳答案

我找不到 linklink_error回调工作,所以我最终不得不使用 on_failureon_success the docs 中描述的任务方法和 this SO question .我的 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)

关于django - 如何在 Celery link_error 回调中获取 "full"异步结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31989366/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com