Alastair
02/16/2026, 6:26 PMMarvin
02/16/2026, 6:26 PMMarvin
02/16/2026, 6:30 PMimport asyncio
import statistics
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import TaskRunFilter, DeploymentFilter
async def task_stats(deployment_name: str, task_name: str, page_size: int = 1000):
async with get_client() as client:
offset = 0
durations = []
while True:
task_runs = await client.read_task_runs(
deployment_filter=DeploymentFilter(name={"any_": [deployment_name]}),
task_run_filter=TaskRunFilter(name={"any_": [task_name]}),
limit=page_size,
offset=offset,
)
if not task_runs:
break
for tr in task_runs:
if tr.start_time and tr.end_time:
durations.append((tr.end_time - tr.start_time).total_seconds())
if len(task_runs) < page_size:
break
offset += page_size
if not durations:
return {
"deployment": deployment_name,
"task": task_name,
"total_runs": 0,
"mean_s": None,
"stdev_s": None,
"min_s": None,
"max_s": None,
}
return {
"deployment": deployment_name,
"task": task_name,
"total_runs": len(durations),
"mean_s": statistics.mean(durations),
"stdev_s": statistics.stdev(durations) if len(durations) > 1 else 0.0,
"min_s": min(durations),
"max_s": max(durations),
}
# Example
if __name__ == "__main__":
print(asyncio.run(task_stats("deployment-x", "task-y")))
Notes and options
- Add a time window if you like by extending the filter, e.g. TaskRunFilter(start_time={"after_": some_dt_utc}).
- You can filter by state as well (e.g., only completed) and keep the “end-start” logic as a safety net.
- For a quick time-series view (counts per bucket and total estimated runtime), Prefect exposes a task run history endpoint. It’s useful for trend charts but still won’t return mean/stdev for you—you’d derive those:
- Server route: server/api/task_runs.py (see /task_runs/history)
- Aggregation logic: server/api/run_history.py
References (source and filters)
- Client method used above: PrefectClient in client.py (method read_task_runs)
- Available filters: prefect.client.schemas.filters
- UI endpoints for counts/time-bucketed views (what powers the UI): server/api/ui/task_runs.py
If you’d like, I can turn this into a small Prefect flow that:
- Computes these stats on a schedule,
- Publishes the results as an artifact or logs,
- Optionally pushes to a warehouse for dashboarding.
What time range and level of granularity do you want (e.g., last 7/30/90 days, per-deployment, per-task)?