https://prefect.io logo
#prefect-community
Title
# prefect-community
m

Milly gupta

04/27/2022, 5:58 PM
Hi All, Is there way to start a new flow using automations? I am looking to trigger a flow B when flow A has finished.
k

Kevin Kho

04/27/2022, 5:59 PM
Yeah! The thread above has an answer here
Did you try this in the UI?
👍 1
a

Anna Geller

04/27/2022, 6:00 PM
it's Automations day 😄 for simple use case you can do that purely from the UI
k

Kevin Kho

04/27/2022, 6:01 PM
I think that might be of the same Flow though? And Milly wants another Flow
a

Anna Geller

04/27/2022, 6:03 PM
wait, Automations is a total overkill here. Why don't you just create a flow of flows using create_flow_run + wait_for_flow_run?
we have plenty of examples here
one example
Copy code
from prefect import Flow
from prefect.storage import GitHub
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
from prefect.run_configs import LocalRun

FLOW_NAME = "04_orchestrating_flow"
STORAGE = GitHub(
    repo="anna-geller/flow-of-flows",
    path=f"flows/{FLOW_NAME}.py",
    access_token_secret="GITHUB_ACCESS_TOKEN",
)
PROJECT_NAME = "jaffle_shop"


with Flow(FLOW_NAME, storage=STORAGE, run_config=LocalRun(labels=["dev"])) as flow:
    extract_load_id = create_flow_run(
        flow_name="01_extract_load",
        project_name=PROJECT_NAME,
        task_args={"name": "Staging"},
    )
    extract_load_wait_task = wait_for_flow_run(
        extract_load_id, raise_final_state=True, task_args={"name": "Staging - wait"}
    )

    transform_id = create_flow_run(
        flow_name="02_dbt",
        project_name=PROJECT_NAME,
        raise_final_state=True,
        task_args={"name": "DBT flow"},
    )
    transform_id_wait_task = wait_for_flow_run(
        transform_id, raise_final_state=True, task_args={"name": "DBT flow - wait"}
    )
    extract_load_wait_task.set_downstream(transform_id)

    dashboards_id = create_flow_run(
        flow_name="03_dashboards",
        project_name=PROJECT_NAME,
        task_args={"name": "Dashboards"},
    )
    dashboards_wait_task = wait_for_flow_run(
        dashboards_id, raise_final_state=True, task_args={"name": "Dashboards - wait"}
    )
    transform_id_wait_task.set_downstream(dashboards_id)
m

Milly gupta

04/27/2022, 9:59 PM
@Anna Geller I appreciate your response but if i wanted to call flow in a flow with the API I would have done that so. If you look at my question I asked about automations.
k

Kevin Kho

04/27/2022, 10:04 PM
The first response here is an automation
m

Milly gupta

04/27/2022, 10:05 PM
Thanks I will look into it
a

Anna Geller

04/27/2022, 11:15 PM
I was genuinely just trying to help 🙂 both Automation and
create_flow_run
work via an API call under the hood
create_flow_run
and setting it in a flow of flows seems a bit easier to configure and easier to troubleshoot
13 Views