conditional-flow.py

from prefect import task, Flow, Parameter
from prefect.tasks.control_flow import case


@task
def send_email(email):
    print(f"Sending an email to {email}")


@task
def send_postcard(planet):
    print(f"Sending a postcard to {planet}")


with Flow("Conditional Example") as flow:

    username = Parameter("username")

    with case(username, "arthur"):
        send_postcard("earth")

    with case(username, "marvin"):
        send_email("[email protected]")


flow.run(username="marvin")  # sending an email to [email protected]
flow.run(username="arthur")  # sending a postcard to earth