Flask Celery update_state из другой функции

Я хотел бы обновить состояние моей задачи Celery из другой функции. Вот что у меня есть сейчас:

Маршрут

@app.route('/my-long-function', methods=['POST'])
def my_long_function():

    param1 = request.form['param1']
    param2 = request.form['param2']

    task = outside_function.delay(param1, param2)

    return task.id

Задача Celery — запускает some_python_script.handle в фоновом режиме.

@celery.task(name='outside_function')
def outside_function(param1, param2):
    with app.app_context():
        some_python_script.handle(param1, param2)

некоторый_python_script.handle:

def handle(param1, param2):
    param1 + param2
    # many, many different things

В идеале я хотел бы иметь возможность self.update_state задачи сельдерея, чтобы я мог легко запросить ее статус из своего приложения, например:

some_python_script.handle (в идеале):

def handle(param1, param2):
    param1 + param2
    # many, many different things
    self.outside_function.update_state('PROGRESS', meta = {'status':'progressing'})

проверить прогресс (в идеале):

@app.route('/status/<task_id>')
def taskstatus(task_id):
    task = outside_function.AsyncResult(task_id)
    response = {
    'state': task.state,
    'id': task.id,
    'status' : task.status,
    }

    return jsonify(response)

Или что-то подобное. Буду признателен за любую помощь, я очень новичок в Celery!


flask celery celery-task
person vandernath    schedule 26.10.2016    source источник

Ответы (2)


arrow_upward
0
arrow_downward

Вы должны объявить идентификатор задачи для вызова. Вы можете проверить update_state .

Приведенный ниже код должен работать.

# capture id of celery task
ID = self.request.id

def handle(param1, param2):
    param1 + param2
    # many, many different things
    # update the state of celery task with direct reference to it
    self.update_state(task_id=ID, state='PROGRESS', meta = {'status':'progressing'})
person amin    schedule 03.01.2017
comment
Это не может работать, потому что в области handle нет self. - person sobek; 21.08.2018

arrow_upward
0
arrow_downward

Если вы добавите bind=True к основной задаче celery, вы получите доступ к объекту задачи celery с ключевым словом self. Передайте объект задачи следующей функции, используя self.

@celery.task(name='outside_function', bind=True)
def outside_function(self, param1, param2):
    with app.app_context():
        some_python_script.handle(self,param1, param2)

Другая функция может принять этот объект задачи и обновить его:

def handle(celery_task, param1, param2):
    param1 + param2
    # many, many different things
    celery_task.update_state('PROGRESS', meta = {'status':'progressing'})
person Mitalee Rao    schedule 02.03.2020