How to access retry attempt in Celery AsyncResult

Question

I use celery with Flask and I have the following task:

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

I have created Flask endpoint to access task results:

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

I would like to return the retry attempt number so that I could display it on the web page. I know that I can do it within the task definition using self.request.retries, but I didn't figure out how to access it from AsyncResult object.

Answer 1

After hours of research, I solved it by adding required information to the exception object. Because this information is not serialized by JSON serializer, I also needed to switch to pickle serializer.

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)