<@ULVA73B9P> is possible to obtain flow run status...
# ask-marvin
h
@Marvin is possible to obtain flow run status and all task status within the flow run in one call with Python SDK
m
thought for 158.5 seconds
Short answer: In Prefect 3.x there isn’t a single high-level Python SDK method that returns the flow run state and all task run states together. The typical approach is two calls: one for the flow run, one for the task runs. If you truly need a single HTTP request, you can call the flow run graph endpoint directly via the client’s low-level request method. Two-call approach (current states only — no per-run state history):
Copy code
from uuid import UUID
from prefect import get_client
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId

async def get_current_states(flow_run_id: UUID):
    async with get_client() as client:
        # 1) Flow run (includes current state)
        flow_run = await client.read_flow_run(flow_run_id)

        # 2) All task runs in this flow run (each includes its current state)
        task_runs = await client.read_task_runs(
            task_run_filter=TaskRunFilter(
                flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])
            )
        )

        return {
            "flow_run_state": flow_run.state,
            "task_run_states": {tr.id: tr.state for tr in task_runs},
        }
If you need state history, call: -
await client.read_flow_run_states(flow_run_id)
for the flow run -
await client.read_task_run_states(task_run_id)
for each task run Single HTTP request option (graph endpoint): - There’s a server endpoint that returns a graph of the flow run including task runs and their state information. There isn’t a dedicated SDK wrapper, but you can use the client’s low-level `request()`:
Copy code
from uuid import UUID
from prefect import get_client

async def get_run_graph(flow_run_id: UUID):
    async with get_client() as client:
        resp = await client.request(
            "GET",
            "/flow_runs/{id}/graph-v2",
            path_params={"id": str(flow_run_id)},
        )
        data = resp.json()
        return data  # includes nodes for the flow run and task runs with state details
Clarifying question: Do you need just the current states, or full state histories for each run? If it’s just current states, the two-call pattern above is the simplest and most efficient. If you must do it in a single HTTP request, the graph endpoint is the closest fit.