Shubham
10/19/2025, 12:53 PMMarvin
10/19/2025, 12:53 PMMarvin
10/19/2025, 12:57 PMimport asyncio
    from prefect.client.orchestration import get_client, wait_for_flow_run
    async def run_and_wait(deployment_id, parameters=None, timeout=3600):
        async with get_client() as client:
            fr = await client.create_flow_run_from_deployment(deployment_id, parameters=parameters or {})
            final = await wait_for_flow_run(fr.id, client=client, timeout=timeout)
            # Retrieve the return value (if it’s trackable)
            result = None
            if final.state and final.state.is_completed():
                result = final.state.result()  # may raise on failure; set raise_on_failure=False if desired
            return final, result
    asyncio.run(run_and_wait("your-deployment-id"))state.result()Shubham
10/19/2025, 1:16 PMMarvin
10/19/2025, 1:16 PMMarvin
10/19/2025, 1:16 PMimport asyncio
from prefect.client.orchestration import get_client, wait_for_flow_run
async def run_deployment_and_get_result(deployment_id: str, parameters: dict | None = None, timeout: int = 3600):
    async with get_client() as client:
        # 1) Create the flow run from the deployment
        fr = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters=parameters or {},
        )
        # 2) Wait for completion
        final = await wait_for_flow_run(fr.id, client=client, timeout=timeout, log_states=True)
        # 3) Fetch the return value (if trackable)
        try:
            value = final.state.result()  # raises if the run failed
            return final, value
        except Exception as exc:
            # Run failed or result retrieval failed
            return final, exc
# Usage:
# final_run, result_or_exc = asyncio.run(run_deployment_and_get_result("your-deployment-id", {"x": 1}))asyncio.run(...)asyncio.runstate.result()flow_run.idstate.result()