How to Implement Conditional Task Execution in Celery from API Registration?

I’m building an API that registers tasks to Celery, and I want to implement a conditional workflow based on the results of the tasks. Specifically, I need to execute Task B if Task A returns a certain result, and I also want to create an on_success callback to execute a different task when Task A completes successfully.

Example Scenario

1. Task A: This task processes some data and returns a result (either True or False).
2. Task B: This task should only execute if Task A returns True.
3. Task C: This task should run after Task A completes successfully, regardless of its result.

Task Definitions

Here’s how I’ve defined my tasks:

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."

API Registration

I want to set up my API in such a way that it registers these tasks with the desired workflow. Here’s what I have in mind for the 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

Desired Workflow

I would like to achieve the following:

• If task_a() returns True, then task_b() should be executed.
• Regardless of the result of task_a(), task_c() should always run after task_a() completes successfully.

My Questions

1. How can I conditionally execute task_b() based on the result of task_a() within the API registration?
2. How can I ensure that task_c() runs after task_a() completes, regardless of its result?
3. What would be the best way to structure this workflow within my API?

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