https://prefect.io
Join Slack
<@ULVA73B9P> how can i generate a report of what is consuming my prefect run hours
# ask-marvin
m
@Marvin how can i generate a report of what is consuming my prefect run hours
m
✅ thought for 151.4 seconds
There isn’t currently a built-in Prefect Cloud report that shows a billing-style breakdown of what is consuming your run hours. What you can do is: * Query flow runs from the API * Sum 
total_run_time
* Group by things like deployment, flow, tags, work pool, or state That’s the best way to generate a usage report today. Recommended approach: use the Python client to build your own report
Copy code
python
from datetime import datetime, timedelta, timezone
import asyncio

from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime, DeploymentFilter, DeploymentFilterId


async def run_hours_report(days_back: int = 30, deployment_ids: list[str] | None = None):
    start = datetime.now(timezone.utc) - timedelta(days=days_back)
    end = datetime.now(timezone.utc)

    flow_run_filter = FlowRunFilter(
        start_time=FlowRunFilterStartTime(after_=start, before_=end)
    )

    deployment_filter = None
    if deployment_ids:
        deployment_filter = DeploymentFilter(
            id=DeploymentFilterId(any_=deployment_ids)
        )

    async with get_client() as client:
        offset = 0
        limit = 200
        runs = []

        while True:
            page = await client.read_flow_runs(
                flow_run_filter=flow_run_filter,
                deployment_filter=deployment_filter,
                limit=limit,
                offset=offset,
            )
            if not page:
                break
            runs.extend(page)
            offset += limit

    total_seconds = 0
    by_flow = {}
    by_deployment = {}

    for run in runs:
        seconds = run.total_run_time.total_seconds() if run.total_run_time else 0
        total_seconds += seconds

        flow_name = run.flow_id
        deployment_name = str(run.deployment_id) if run.deployment_id else "no deployment"

        by_flow[flow_name] = by_flow.get(flow_name, 0) + seconds
        by_deployment[deployment_name] = by_deployment.get(deployment_name, 0) + seconds

    print(f"Period: {start.isoformat()} -> {end.isoformat()}")
    print(f"Total flow runs: {len(runs)}")
    print(f"Total run hours: {total_seconds / 3600:.2f}")

    print("\nTop deployments by run hours:")
    for name, seconds in sorted(by_deployment.items(), key=lambda x: x[1], reverse=True)[:10]:
        print(f"  {name}: {seconds / 3600:.2f}h")

    print("\nTop flows by run hours:")
    for name, seconds in sorted(by_flow.items(), key=lambda x: x[1], reverse=True)[:10]:
        print(f"  {name}: {seconds / 3600:.2f}h")


if __name__ == "__main__":
    asyncio.run(run_hours_report(days_back=30))
A few notes: * 
total_run_time
is the key field to inspect * You can filter further by deployment, flow, tags, dates, or work pools * If you want a real billing approximation, multiply the resulting hours by your plan’s managed compute pricing from prefect.io/pricing If you’re specifically using Prefect Managed Execution, there are also usage-related events you can inspect, but they’re not as straightforward as a ready-made report. Useful links: * Prefect pricing: prefect.io/pricing * Prefect Cloud app: app.prefect.cloud If you want, I can help you turn this into: * a CSV report * a per-deployment / per-work-pool breakdown * or a scheduled monthly usage report flow
m
@Marvin Generate a csv
m
✅ thought for 19.8 seconds
Absolutely — here’s a simple script that exports your Prefect flow run usage to a CSV. It will generate a file with one row per flow run and include: * flow run ID * flow name * deployment ID * run state * start/end timestamps * total run time in seconds * total run time in hours
Copy code
python
import csv
import asyncio
from datetime import datetime, timedelta, timezone

from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime


async def export_run_hours_csv(days_back: int = 30, output_file: str = "prefect_run_hours_report.csv"):
    start = datetime.now(timezone.utc) - timedelta(days=days_back)
    end = datetime.now(timezone.utc)

    flow_run_filter = FlowRunFilter(
        start_time=FlowRunFilterStartTime(after_=start, before_=end)
    )

    rows = []

    async with get_client() as client:
        offset = 0
        limit = 200

        while True:
            flow_runs = await client.read_flow_runs(
                flow_run_filter=flow_run_filter,
                limit=limit,
                offset=offset,
            )

            if not flow_runs:
                break

            for run in flow_runs:
                total_seconds = run.total_run_time.total_seconds() if run.total_run_time else 0

                rows.append({
                    "flow_run_id": str(run.id),
                    "flow_name": getattr(run, "flow_name", "") or "",
                    "deployment_id": str(run.deployment_id) if run.deployment_id else "",
                    "deployment_name": getattr(run, "deployment_name", "") or "",
                    "state_name": run.state_name or "",
                    "start_time": run.start_time.isoformat() if run.start_time else "",
                    "end_time": run.end_time.isoformat() if run.end_time else "",
                    "total_run_time_seconds": round(total_seconds, 2),
                    "total_run_time_hours": round(total_seconds / 3600, 4),
                })

            offset += limit

    with open(output_file, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=[
                "flow_run_id",
                "flow_name",
                "deployment_id",
                "deployment_name",
                "state_name",
                "start_time",
                "end_time",
                "total_run_time_seconds",
                "total_run_time_hours",
            ],
        )
        writer.writeheader()
        writer.writerows(rows)

    print(f"Wrote {len(rows)} rows to {output_file}")


if __name__ == "__main__":
    asyncio.run(export_run_hours_csv(days_back=30))
To run it:
Copy code
bash
python export_prefect_run_hours.py
If you want a quick summary from that CSV afterward, you can use pandas:
Copy code
python
import pandas as pd

df = pd.read_csv("prefect_run_hours_report.csv")
print(df.groupby("deployment_name")["total_run_time_hours"].sum().sort_values(ascending=False))
If you want, I can also give you a version that: * groups directly into a per-deployment summary CSV * includes work pool / tags * or exports only managed execution runs
m
@Marvin This does not add up to the total number of hours i have been billed for
m
✅ thought for 141.4 seconds
Yep — that makes sense. The main reason is: 
total_run_time
is not the same thing as billed managed compute time. For Prefect Cloud managed execution, billing is closer to: compute startup → compute teardown, rounded up to the nearest minute not just: time spent actually running user code So if you summed 
run.total_run_time
, you will usually under-count billed hours. What causes the mismatch 1. PENDING time is billed * A run can spend time in 
PENDING
while infrastructure starts up * 
total_run_time
does not include that 2. Billing rounds up to the nearest minute * A 10-second run may still bill as 1 minute * Lots of short runs can create a big difference 3. Startup/teardown overhead is billed * Container spin-up, image prep, shutdown, etc. 4. Retries add overhead each time * Each attempt may incur its own startup + rounded minute billing 5. Failed/cancelled runs can still consume billed time * Even if they don’t run long So if you want a closer estimate of billed run hours, you should use: * 
start_time
* 
end_time
* round each run up to the nearest minute Here’s a CSV generator that does that instead. ```python import csv import math import asyncio from datetime import datetime, timedelta, timezone from prefect import get_client from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime async def export_billed_hours_csv( days_back: int = 30, output_file: str = "prefect_billed_hours_estimate.csv", ): start = datetime.now(timezone.utc) - timedelta(days=days_back) end = datetime.now(timezone.utc) flow_run_filter = FlowRunFilter( start_time=FlowRunFilterStartTime(after_=start, before_=end) ) rows = [] async with get_client() as client: offset = 0 limit = 200 while True: flow_runs = await client.read_flow_runs( flow_run_filter=flow_run_filter, limit=limit, offset=offset, ) if not flow_runs: break for run in flow_runs: actual_seconds = run.total_run_time.total_seconds() if run.total_run_time else 0 billed_seconds_estimate = 0 billed_minutes_estimate = 0 if run.start_time and run.end_time: wall_clock_seconds = max((run.end_time - run.start_time).total_seconds(), 0) billed_minutes_estimate = math.ceil(wall_clock_seconds / 60) billed_seconds_estimate = billed_minutes_estimate * 60 else: wall_clock_seconds = 0 rows.append({ "flow_run_id": str(run.id), "flow_name": getattr(run, "flow_name", "") or "", "deployment_id": str(run.deployment_id) if run.deployment_id else "", "deployment_name": getattr(run, "deployment_name", "") or "", "state_name": run.state_name or "", "start_time": run.start_time.isoformat() if run.start_time else "", "end_time": run.end_time.isoformat() if run.end_time else "", "actual_run_time_seconds": round(actual_seconds, 2), "actual_run_time_hours": round(actual_seconds / 3600, 4), "wall_clock_seconds": round(wall_clock_seconds, 2), "estimated_billed_minutes": billed_minutes_estimate, "estimated_billed_hours": round(billed_seconds_estimate / 3600, 4), }) offset += limit with open(output_file, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( f, fieldnames=[ "flow_run_id", "flow_name", "deployment_id", "deployment_name", "state_name", "start_time",
"end_time", "actual_run_time_seconds", "actual_run_time_hours", "wall_clock_seconds", "estimated_billed_minutes", "estimated_billed_hours", ], ) writer.writeheader() writer.writerows(rows) total_estimated_hours = sum(row["estimated_billed_hours"] for row in rows) total_actual_hours = sum(row["actual_run_time_hours"] for row in rows) print(f"Wrote {len(rows)} rows to {output_file}") print(f"Total actual run hours: {total_actual_hours:.4f}") print(f"Total estimated billed hours: {total_estimated_hours:.4f}") if name == "__main__": asyncio.run(export_billed_hours_csv(days_back=30)) ``` Important caveat This is still an estimate. Prefect’s docs describe managed compute usage as approximately the duration from 
PENDING
to terminal state, rounded up to the nearest minute. Depending on exactly how your runs are executed, the invoice may still differ somewhat. If you want, I can give you a better version that exports two CSVs: * 
run_level_detail.csv
* 
deployment_summary.csv
so you can see which deployments are driving billed hours.
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by