Как получить расчетное оставшееся время уже запущенной задачи Celery?

Я использую django с Celery и redis. Я хотел бы, чтобы, когда пользователь инициирует задачу Celery, мое приложение оценивало количество времени, оставшееся до завершения задачи (чтобы я мог, например, представить эти данные на фронтенде в виде прогресс-бара).

Например, данная функция опроса:

views.py:

@csrf_protect
def initiate_task(request):
    task = new_celery_task.delay(parmeter_1, parametr_2)

@csrf_protect
def poll_task_status(request, task_id): #task_id provided by frontend
    
    import base64
    task_result = AsyncResult(task_id)
    
    if task_result.ready():
        try:
            ...
                    
                    response_data = {
                        'status': 'completed',
                        }
        
        except Task.DoesNotExist:
            return JsonResponse({'status': 'error', 'message': 'There has been an error'}, status=404)
    
    elif task_result.state == 'REVOKED':
        return JsonResponse({'status': 'terminated', 'message': 'Task was terminated'})
    
    else:
        #add code to calculate estimated remaining time here
        remaining_time_est = estimate_remaining_time() #estimate task by id
        return JsonResponse({'status': 'pending'
                             #add code to return estimated remaining time here
                             'est_remaining_time':remaining_time_est
                             })

и задача:

task.py:

app = Celery('myapp')#
@app.task 
def new_celery_task(arg_1, arg_2):
    #complete task

Как оценить оставшееся время выполнения начатого задания?

Вернуться на верх