import asyncio

from prefect import get_client
from prefect.flow_runs import wait_for_flow_run

async def main():
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(deployment_id="my-deployment-id")
        flow_run = await wait_for_flow_run(flow_run_id=flow_run.id)
        print(flow_run.state)

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

@task
def task_1():
    pass

@task
def task_2():
    pass

@flow
def my_flow():
    x = task_1()

    # task 2 will wait for task_1 to complete
    y = task_2(wait_for=[x])