Как реализовать условное выполнение задачи в Celery из регистрации API? ⇐ Python

[Anonymous]

Я создаю API, который регистрирует задачи в Celery, и хочу реализовать условный рабочий процесс на основе результатов выполнения задач. В частности, мне нужно выполнить задачу Б, если задача А возвращает определенный результат, а также я хочу создать обратный вызов on_success для выполнения другой задачи при успешном завершении задачи А.

Пример сценария

• Задача A: эта задача обрабатывает некоторые данные и возвращает результат (либо True, либо False< /code>).
• Задача B: эта задача должна выполняться только в том случае, если задача A возвращает значение True.
  < li>Задача C. Эта задача должна выполняться после успешного завершения задачи A, независимо от ее результата.

Определения задачЗадача C. h4>
Вот как я определил свои задачи:
[code]from celery import Celery, chain

app = Celery('my_tasks', broker='redis://localhost:6379/0')

@app.task
def task_a():
# Simulate processing and returning a result
result = True # or False, based on some condition
return result

@app.task
def task_b():
return "Task B executed."

@app.task
def task_c():
return "Task C executed."
[/code]
Регистрация API

Я хочу настроить свой API таким образом, чтобы он регистрировал эти задачи в желаемом рабочем процессе. Вот что я имею в виду в отношении API:

Код: Выделить всё

from flask import Flask, request
from my_tasks import task_a, task_b, task_c

app = Flask(__name__)

@app.route('/register-task', methods=['POST'])
def register_task():
# Register task_a and create a conditional workflow
workflow = (
task_a.s()  # Start with task_a
| task_b.s()  # This should execute if task_a returns True
| task_c.s()  # This should always run after task_a
)

# Call the workflow
result = workflow.delay()  # Queue the workflow

return {"task_id": result.id}, 202

Желаемый рабочий процесс

Я хотел бы добиться следующего:

• Если Task_a() возвращает True, то Task_b() должен быть выполнен.
• Независимо от результата Task_a() , Task_c() всегда должен запускаться после успешного завершения Task_a().

Мои вопросы

• Как я могу условно выполнить Task_b() на основе результата Task_a() при регистрации API?
• Как я могу гарантировать, что Task_c() запускается после завершения Task_a(), независимо от его результата?
• Как лучше всего структурировать этот рабочий процесс в моем API?

............ .................................................. .........

Подробнее здесь: https://stackoverflow.com/questions/791 ... gistration