Yes — there’s definitely prior art, but mostly via the Python client / REST API, not the CLI. Short version: - Historical event queries: supported via the client library / API - CLI: supports streaming events, not really “download all historical events for analysis” - For your use case, the best path is usually: 1. query task-run events with

read_events(...)

2. filter to your flow run(s) 3. compute metrics from event timestamps and/or task-run metadata A few important notes for Prefect 3.x: What exists today - Python client supports querying events - REST API supports filtering/pagination of events - CLI does not currently have a solid “export historical events” workflow I wasn’t able to confirm CLI subcommands from

--help

because the CLI check timed out, so I’d avoid recommending a CLI-based solution here. Recommended approach: Python client You can query

prefect.task-run.Completed

events like this:

python
import asyncio
from datetime import timedelta

from prefect import get_client
from prefect.events.filters import (
    EventFilter,
    EventNameFilter,
    EventOccurredFilter,
)
import prefect.types._datetime


async def fetch_completed_task_events(flow_run_id: str):
    now = prefect.types._datetime.now("UTC")

    event_filter = EventFilter(
        occurred=EventOccurredFilter(
            since=now - timedelta(days=7),
            until=now,
        ),
        event=EventNameFilter(
            name=["prefect.task-run.Completed"]
        ),
    )

    matched = []

    async with get_client() as client:
        page = await client.read_events(filter=event_filter, limit=100)

        while page:
            for event in page.events:
                task_run = event.payload.get("task_run", {})
                if str(task_run.get("flow_run_id")) == flow_run_id:
                    matched.append(event)

            page = await page.get_next_page(client) if page.next_page else None

    return matched


async def main():
    flow_run_id = "YOUR-FLOW-RUN-ID"
    events = await fetch_completed_task_events(flow_run_id)

    for event in events:
        task_run = event.payload.get("task_run", {})
        print({
            "task_run_id": task_run.get("id"),
            "task_name": task_run.get("name"),
            "completed_at": str(event.occurred),
            "total_run_time": task_run.get("total_run_time"),
        })


asyncio.run(main())

What you can extract from the event For

prefect.task-run.Completed

, useful fields typically include: -

event.occurred

-

event.resource

-

event.payload["task_run"]

-

event.payload["validated_state"]

In particular,

event.payload["task_run"]["total_run_time"]

is likely the easiest duration metric if present in your environment. One caveat Task-run completion events are easy to filter by event name, but to scope them to a specific flow run, you’ll often end up filtering in Python using:

python
event.payload.get("task_run", {}).get("flow_run_id")

That’s the simplest pattern if you want “all completed tasks in this flow run”. If you want timing metrics beyond total runtime You can compute: - per-task runtime:

task_run["total_run_time"]

- completion timestamp:

event.occurred

- queue-to-start / start-to-finish style metrics: often easier to derive by combining task run objects with state transition events, depending on how exact you need to be If your goal is just runtime summaries, you may actually find the task run API simpler than events. For example: - query task runs for a flow run - inspect task-run timestamps / state / total run time - use events only if you specifically want state-transition timing When events are the right tool Use events if you want: - state transition audit trail - precise transition chronology - event-driven analytics across many runs Use task-run records if you want: - straightforward duration summaries - fewer pagination/filtering steps API path

Under the hood, this is backed by the Events API, including filtering/pagination endpoints. If you want, I can put together either: 1. a complete script that exports all

prefect.task-run.Completed

events for a given

flow_run_id

to CSV, or 2. a version that compares task-run API vs events API for computing timing metrics.