如何在Celery AsyncResult中访问重试尝试 - python

我将芹菜与Flask一起使用，并且执行以下任务：

@celery.task(bind=True, max_retries=5)
def failing_task(self):
    time.sleep(random.randint(4, 10))
    try:
        raise ValueError('Exception occurred')
    except Exception as e:
        self.retry(exc=e, countdown=int(random.uniform(2, 3) ** self.request.retries))

我创建了Flask端点来访问任务结果：

@app.route('/api/task/<task_id>')
def task_status(task_id):
    task = AsyncResult(task_id)
    task_result = task.result
    if isinstance(task_result, Exception):
        task_result = '{e.__class__.__name__}: {e}'.format(e=task_result)

    app.logger.info(task.info)
    content = {'task_id': task.id,
               'state': task.state,
               'result': task_result}

    return jsonify(content)

我想返回重试次数，以便可以在网页上显示它。我知道我可以使用self.request.retries在任务定义中完成此操作，但是我不知道如何从AsyncResult对象访问它。

参考方案

经过数小时的研究，我通过将所需信息添加到异常对象中来解决了该问题。因为此信息未由JSON序列化程序序列化，所以我还需要切换到pickle序列化程序。

app/tasks.py

from celery import Celery

celery = Celery('tasks', broker=constants.CELERY_URL, backend=constants.CELERY_URL)
celery.conf.CELERYD_TASK_SOFT_TIME_LIMIT = 120
celery.conf.CELERY_RESULT_SERIALIZER = 'pickle'
celery.conf.CELERY_TASK_SERIALIZER = 'pickle'
celery.conf.CELERY_ACCEPT_CONTENT = ['json', 'pickle']

@celery.task(bind=True)
def failing_task(self):
    try:
        time.sleep(random.randint(4, 10))
        raise ValueError('Exception occurred in failing task')
    except Exception as e:
        e.retry_number = self.request.retries + 1
        e.max_retries = self.max_retries
        self.retry(exc=e, countdown=int(random.uniform(2, 3) ** self.request.retries))

app/main.py

@app.route('/api/task/<task_id>')
def task_status(task_id):
    task = AsyncResult(task_id)
    content = {'task_id': task.id,
               'state': task.state,
               'result': task.result}

    if isinstance(task.result, Exception):
        content['result'] = '{e.__class__.__name__}: {e}'.format(e=task.result)
        content['retry_number'] = task.result.retry_number
        content['max_retries'] = task.result.max_retries

    return jsonify(content)

在返回'Response'(Python)中传递多个参数 - python

我在Angular工作，正在使用Http请求和响应。是否可以在“响应”中发送多个参数。角度文件：this.http.get("api/agent/applicationaware").subscribe((data:any)... python文件：def get(request): ... return Response(seriali…

R'relaimpo'软件包的Python端口 - python

我需要计算Lindeman-Merenda-Gold（LMG）分数，以进行回归分析。我发现R语言的relaimpo包下有该文件。不幸的是，我对R没有任何经验。我检查了互联网，但找不到。这个程序包有python端口吗？如果不存在，是否可以通过python使用该包？ python参考方案 最近，我遇到了pingouin库。

如何用'-'解析字符串到节点js本地脚本？ - python

我正在使用本地节点js脚本来处理字符串。我陷入了将'-'字符串解析为本地节点js脚本的问题。render.js：#! /usr/bin/env -S node -r esm let argv = require('yargs') .usage('$0 [string]') .argv; console.log(argv…

TypeError:'str'对象不支持项目分配，带有json文件的python - python

以下是我的代码import json with open('johns.json', 'r') as q: l = q.read() data = json.loads(l) data['john'] = '{}' data['john']['use…

Python ThreadPoolExecutor抑制异常 - python

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED def div_zero(x): print('In div_zero') return x / 0 with ThreadPoolExecutor(max_workers=4) as execut…