I have two flows deployed on two distinct aws accounts that I'd like to coordinate as subflows. Right now I'm calling them sequentially with something like

my_flow_run = prefect.deployments.run_deployment(name='my_deployment')

, then throwing an error if

my_flow_run.state_name=='failed'

. I'd like to coordinate them asyncronously, with something like

my_flow_run_1, my_flow_run_2 = await asyncio.gather(
        run_deployment(
            name=f"my_deployment_1"),
        run_deployment(
            name=f"my_deployment_2")
    )
throw_error_if_flow_run_failed(my_flow_run_1)
throw_error_if_flow_run_failed(my_flow_run_2)

then I'd like to continue the rest of the main flow sequentially. Am I going about this the right way? is there a built-in solution or will I need to hand-roll something?

Prefect out of the box handles concurrent execution, so you can have the logic as tasks in a flow, and log errors whenever they get thrown. Then on the infra side you can set a concurrency limit if you want to add a ceiling on the computation

@Max Jackson this might actually be what you are looking for. I have started implementing a similar pattern to this. It works nicely with pythons

async

keywords. https://docs.prefect.io/latest/tutorials/execution/#asynchronous-execution

@Nelson Griffiths this is great and exactly what I have in mind, but I'm not sure that I can do this without an

async

definition of

run_deployment

. namely, I keep getting

AttributeError: 'coroutine' object has no attribute 'state_type'

It sounds like you aren’t awaiting your gather. Could you share your code snippet?