Great question. Assuming Prefect 3.x, here’s a simple pattern to: - list all deployments - fetch their flow runs from the last N days - compute average run time (by default for completed runs only) Async example (recommended)

import asyncio
from datetime import datetime, timezone, timedelta

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    DeploymentFilter,
    DeploymentFilterId,
    FlowRunFilter,
    FlowRunFilterStartTime,
    FlowRunFilterState,
    FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType

async def avg_run_time_by_deployment(days: int = 7, include_failed: bool = False):
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    async with get_client() as client:
        deployments = await client.read_deployments()

        results = []
        for dep in deployments:
            # Filter runs in the window; by default, only COMPLETED runs
            state_types = [StateType.COMPLETED]
            if include_failed:
                state_types.append(StateType.FAILED)

            flow_run_filter = FlowRunFilter(
                start_time=FlowRunFilterStartTime(after_=cutoff),
                state=FlowRunFilterState(
                    type=FlowRunFilterStateType(any_=state_types)
                ),
            )

            runs = await client.read_flow_runs(
                deployment_filter=DeploymentFilter(
                    id=DeploymentFilterId(any_=[dep.id])
                ),
                flow_run_filter=flow_run_filter,
                # bump this or paginate if you have lots of runs
                limit=1000,
            )

            durations = [r.total_run_time.total_seconds() for r in runs]
            avg_seconds = (sum(durations) / len(durations)) if durations else 0.0

            # Many users find flow_name/deployment_name helpful for readability
            dep_label = f"{getattr(dep, 'flow_name', 'flow')}/{dep.name}"

            results.append({
                "deployment_id": dep.id,
                "deployment_name": dep_label,
                "run_count": len(durations),
                "avg_seconds": avg_seconds,
            })

        return results


if __name__ == "__main__":
    rows = asyncio.run(avg_run_time_by_deployment(days=7))
    for r in rows:
        print(f"{r['deployment_name']}: {r['run_count']} runs, avg {r['avg_seconds']:.1f}s")

Sync variant ``` from datetime import datetime, timezone, timedelta from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( DeploymentFilter, DeploymentFilterId, FlowRunFilter, FlowRunFilterStartTime, FlowRunFilterState, FlowRunFilterStateType, ) from prefect.client.schemas.objects import StateType def avg_run_time_by_deployment_sync(days: int = 7, include_failed: bool = False): cutoff = datetime.now(timezone.utc) - timedelta(days=days) with get_client(sync_client=True) as client: deployments = client.read_deployments() results = [] for dep in deployments: state_types = [StateType.COMPLETED] if include_failed: state_types.append(StateType.FAILED) flow_run_filter = FlowRunFilter( start_time=FlowRunFilterStartTime(after_=cutoff), state=FlowRunFilterState( type=FlowRunFilterStateType(any_=state_types) ), ) runs = client.read_flow_runs( deployment_filter=DeploymentFilter( id=DeploymentFilterId(any_=[dep.id]) ), flow_run_filter=flow_run_filter, limit=1000, ) durations = [r.total_run_time.total_seconds() for r in runs] avg_seconds = (sum(durations) / len(durations)) if durations else 0.0 dep_label = f"{getattr(dep, 'flow_name', 'flow')}/{dep.name}"

results.append((dep_label, len(durations), avg_seconds)) return results if name == "__main__": for name, count, avg_s in avg_run_time_by_deployment_sync(days=7): print(f"{name}: {count} runs, avg {avg_s:.1f}s") ``` Notes - We filter by

start_time

in the last N days and compute the average from

FlowRun.total_run_time

. - By default, the example averages only COMPLETED runs; pass

include_failed=True

to include failed runs as well. - If you expect a large number of runs, increase

limit

or paginate with

offset

to collect all runs before averaging. Would you like the average to include failed/cancelled runs, or only completed ones? And do you prefer the async or sync version?