from prefect.engine import FlowRun
from prefect.server.schemas.responses import DeploymentResponse
from prefect.states import Scheduled, State

async def trigger_flow_deployment(flow: Flow, flow_run: FlowRun, state: State):
    """Schedules a flow to run immediately. To be used with a Flow's `on_completion`."""
    result = await state.result(fetch=True)
    async with get_client() as client:

        deployment: DeploymentResponse = await get_deployment(
            client, name="My Flow/Deployment Name", deployment_id=None
        )

        await client.create_flow_run_from_deployment(
            deployment.id,
            parameters=...,
            state=Scheduled(),
        )

run_A

on_completion

import pendulum
from prefect import flow
from prefect.deployments import Deployment, run_deployment
from prefect.server.schemas.schedules import CronSchedule


# refs:
#   <https://github.com/PrefectHQ/prefect-recipes/blob/main/flows-advanced/parent-orchestrator/pokemon_weight.py#L57-L65>
#   <https://github.com/PrefectHQ/prefect/blob/6065be2434871ca1164d85ca00d9cb5bce509445/src/prefect/deployments/deployments.py#L48>

# deploy this flow with:
# prefect deployment build test_deploy_dep_flows.py:run_upstream -n upstream_deployment -a
@flow(log_prints=True,)
async def run_upstream():
    from time import sleep
    print(f"running upstream @ {pendulum.now(tz='America/New_York')}")
    sleep(2*60)
    print(f"...upstream flow completed @ {pendulum.now(tz='America/New_York')}")


# deploy this flow with:
# prefect deployment build test_deploy_dep_flows.py:run_downstream -n downstream_deployment -a
@flow(log_prints=True)
async def run_downstream():
    await run_deployment(name='run-upstream/upstream_deployment')
    print(f"running downstream @ {pendulum.now(tz='America/New_York')}...")


if __name__ == '__main__':
    deployment = Deployment.build_from_flow(
        flow=run_upstream,
        name="upstream_deployment",
        version=1,
        work_queue_name="default",
        work_pool_name="default-agent-pool",
        schedule=CronSchedule(cron="37 20 * * 1", timezone="America/New_York"),
    )
    deployment.apply()

    deployment = Deployment.build_from_flow(
        flow=run_downstream,
        name="downstream_deployment",
        version=1,
        work_queue_name="default",
        work_pool_name="default-agent-pool",
        schedule=CronSchedule(cron="38 20 * * 1", timezone="America/New_York"),
    )
    deployment.apply()

run_downstream

run_upstream