Want to run flow when another project flow-run is completed

[arika0093]

Is there any way to make the following situation work with Prefect v2 ?
It seems like a common use case, but I couldn't find any similar cases when I looked into it.

# sample project
# A: collect datas from some files and save them to something.
from prefect import flow

@flow(name="A")
def collect_and_save():
  data_a = fetch_some_data()
  data_b = fetch_other_data()
  data_save(data_a, data_b)

# other project
# B: detects that A is complete and performs an analysis from that file.
from prefect import flow

@flow(name="B")
def analyse_datas():
  data_a, data_b = get_datas()
  data_c = some_analyse(data_a, data_b)
  write_something(data_c)

# with Flow("A").when_completed():
#   analyse_datas()

For reference, the following page was explained in Prefect v1 . Exactly what we want to do.
https://docs-v1.prefect.io/core/idioms/flow-to-flow.html#scheduling-a-flow-of-flows

It doesn't matter if it's in the form of stating the dependencies in the code or specifying them in the web UI.

Thanks in advance.

[aliaramli]

https://docs.prefect.io/concepts/flows/#composing-flows

try to use wait_for parameter in Prefect 2, I haven't test though

[arika0093]

Thanks for answering.
I tried briefly, but it does not work due to two things.

1: The wait_for parameter can only be used with Task; trying to use it with Flow results in AttributeError: 'Flow' object has no attribute 'submit'.
2: We are assuming a case where the function entity cannot be referred to in the first place.

On second thought, I don't want to wait for Flow to complete, I want to wait for Flow-Run to complete.
so I changed the question statement to.

# not like this
@flow
def flowA():
  pass

@flow
def flowB():
  pass

flowB.submit(wait_for=flowA)

# like this

# @flow(name="flow_A")
# def flowA()  # <- runs in a separate deployments.

@flow
def flowB():
  pass

flowB.submit(wait_for=Flow("flow_A"))

[galtman5]

It seems like you can use the recently introduced automations. In the cloud UI, on the left menu, there is an automations tab where you can build a config that makes one flow trigger when another flow hits a state of complete.

[arika0093]

Exactly what I wanted to do is there!
However, I am a little disappointed that it is cloud only. I wish it could be done self-hosting as well.
Anyway, thanks for the answer.

[leandro-ferreira-farm]

I have the same problem, and I like to do this in self-hosting server and using Python programming. I saw the @arika0093 could do this using additional with Orion API, but I think there should be an easier way and do this, for example: sending a sign of to the next I will be waiting for new open source versions to have this feature available.

[ahuang11]

Disregarding "We are assuming a case where the function entity cannot be referred to in the first place" for a moment, I think this works:

def a_flow():
    print("a")

def b_flow():
    print("b")

def parent_flow():
    a_flow()
    b_flow()

parent_flow()

[zachloertscher]

I know this is an old thread, but I'm having the same question

The above works if all of your sub-flows are all running on the same infrastructure, but what about when you have flow_a running on an EC2, and flow_b & flow_c are running on an ECS cluster? I'm not aware that we could do that with subflows.

I'm finding that my only options are a hacky work around using the Orion API, or using an Automation in the Cloud. Unless anyone has addressed this?

[arika0093]

I was able to forcefully implement this using the Orion API.
I leave it as a note for those who are curious about this topic.

import asyncio
from time import sleep
from datetime import datetime
from prefect import flow
from prefect.client.orion import get_client
from prefect.orion.schemas.filters import (
  FlowRunFilterState,
  FlowRunFilterStateName ,
  FlowRunFilter,
  FlowRunFilterStartTime,
  DeploymentFilter,
  DeploymentFilterName)

async def wait_flow_end(
  deploy_name: str,
  start_time = datetime.now(),
  max_loop_time: int = 3600,
  retry_span: int = 10
):
  current_t = 0
  client = get_client()
  while current_t < max_loop_time:
    states = await client.read_flow_runs(
      flow_run_filter=FlowRunFilter(
        state=FlowRunFilterState(
          type=FlowRunFilterStateName(any_=["COMPLETED"])
        ),
        start_time=FlowRunFilterStartTime(
          after_=start_time
        )
      ),
      deployment_filter=DeploymentFilter(
        name=DeploymentFilterName(any_=[deploy_name])
      ),
      limit=1
    )
    if(len(states) > 0):
      print(f"flow-run completed!")
      return True

    current_t += retry_span
    print(f"wait flow-run complete of deploy:{deploy_name} ({current_t}/{max_loop_time} sec)")
    sleep(retry_span)

  return False


@flow(name="test_B")
async def testFlow_B():
  print("Hello, Flow B!")

@flow
async def testFlow_runB_afterA():
  if(await wait_flow_end("test_A")):
    await testFlow_B()


if __name__ == "__main__":
  asyncio.run( testFlow_runB_afterA() )

The output is then as follows. (I have an Orion server running on the same host, with a deployment named test_A. )

❯ python flow_B.py 
15:30:20.468 | INFO    | prefect.engine - Created flow run 'esoteric-alpaca' for flow 'testFlow-runB-afterA'
wait flow-run complete of deploy:test_A (10/3600 sec)
wait flow-run complete of deploy:test_A (20/3600 sec)
15:30:39.671 | INFO    | prefect.agent - Submitting flow run 'ed0fe644-467d-4a5a-abdd-bc01290bd680'
15:30:39.732 | INFO    | prefect.infrastructure.process - Opening process 'didactic-herring'...
15:30:39.748 | INFO    | prefect.agent - Completed submission of flow run 'ed0fe644-467d-4a5a-abdd-bc01290bd680'
wait flow-run complete of deploy:test_A (30/3600 sec)
15:30:41.689 | INFO    | Flow run 'didactic-herring' - Downloading flow code from storage at '/workspaces/python'
15:30:41.844 | INFO    | Flow run 'didactic-herring' - Finished in state Completed()
Hello, Flow A!
15:30:42.189 | INFO    | prefect.infrastructure.process - Process 'didactic-herring' exited cleanly.
flow-run completed!
15:30:50.704 | INFO    | Flow run 'esoteric-alpaca' - Created subflow run 'opalescent-millipede' for flow 'test_B'
Hello, Flow B!
15:30:50.782 | INFO    | Flow run 'opalescent-millipede' - Finished in state Completed()
15:30:50.808 | INFO    | Flow run 'esoteric-alpaca' - Finished in state Completed('All states completed.')