Great question. Quick clarifier so I can tailor the steps: - Are you on Prefect Cloud or self-hosted Prefect Server? - By “flows” do you mean “flow runs”? (Flows are code definitions; typically you want to delete flow runs, logs, and artifacts older than 30 days.) Assuming Prefect 3.x and that you want to purge items older than 30 days, here’s a pragmatic way to do it with a small Prefect cleanup flow you can schedule to run periodically. What this does - Finds flow runs that started more than 30 days ago - Deletes their artifacts - Deletes the flow runs (which removes their run records and, in practice, their logs) - You can optionally call the “delete flow-run logs” endpoint first if you want to aggressively prune logs Cleanup flow (ready to run)

from datetime import datetime, timedelta, timezone

from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterStartTime,
    ArtifactFilter,
    ArtifactFilterFlowRunId,
)

CUTOFF_DAYS = 30
PAGE_SIZE = 200

@flow
async def purge_old_history(days: int = CUTOFF_DAYS):
    logger = get_run_logger()
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    <http://logger.info|logger.info>(f"Purging artifacts, logs, and flow runs older than {days} days (before {cutoff.isoformat()})")

    async with get_client() as client:
        # Build filter for old flow runs
        fr_filter = FlowRunFilter(start_time=FlowRunFilterStartTime(before_=cutoff))

        total_deleted = 0
        offset = 0

        while True:
            # Page through old flow runs
            runs = await client.read_flow_runs(
                limit=PAGE_SIZE,
                offset=offset,
                flow_run_filter=fr_filter,
            )
            if not runs:
                break

            for fr in runs:
                # 1) Delete artifacts tied to this flow run
                af = ArtifactFilter(flow_run_id=ArtifactFilterFlowRunId(any_=[fr.id]))
                arts_offset = 0
                while True:
                    arts = await client.read_artifacts(limit=PAGE_SIZE, offset=arts_offset, artifact_filter=af)
                    if not arts:
                        break
                    for art in arts:
                        try:
                            await client.delete_artifact(art.id)
                        except Exception as e:
                            logger.warning(f"Failed to delete artifact {art.id} for flow run {fr.id}: {e}")
                    arts_offset += len(arts)

                # 2) Optionally, delete logs for this flow run first (extra-aggressive pruning)
                try:
                    if hasattr(client, "delete_flow_run_logs"):
                        await client.delete_flow_run_logs(fr.id)
                except Exception as e:
                    logger.warning(f"Failed to delete logs for flow run {fr.id}: {e}")

                # 3) Delete the flow run
                try:
                    await client.delete_flow_run(fr.id)
                    total_deleted += 1
                except Exception as e:
                    logger.warning(f"Failed to delete flow run {fr.id}: {e}")

            offset += len(runs)

        <http://logger.info|logger.info>(f"Deleted {total_deleted} flow runs (and associated artifacts).")
        return total_deleted

if __name__ == "__main__":
    purge_old_history()

Notes and options - Artifacts: Prefect’s CLI supports deleting a single artifact by key or id: - List:

prefect artifact ls

- Delete by key:

prefect artifact delete "my-artifact-key"

- Delete by id:

prefect artifact delete --id <UUID>

- Docs: CLI: artifact

- Logs: There is a flow-run-scoped logs delete endpoint; this code attempts to call it if present. In practice, deleting the flow run removes its associated execution records; calling the logs endpoint first just ensures logs are pruned even if flow-run deletion fails. - Flow run delete: REST: DELETE /flow_runs/{id} - Flow-run logs delete (scoped): documented via server API functions - Cloud vs self-hosted: - Prefect Cloud has built-in retention for some data (varies by tier). If you only want to enforce “30 days” and your workspace retention is shorter, Cloud will prune automatically. If you want earlier pruning, running the cleanup flow above is fine. - Self-hosted: There is an events retention setting (

PREFECT_SERVER_EVENTS_RETENTION_PERIOD

) but it only covers events, not logs/artifacts/flow runs. See Database maintenance. - Schedule it: You can deploy this cleanup flow and schedule it to run nightly/weekly. In Prefect 3.x, use

flow.from_source(...).deploy(...)

if deploying from a repo, or run it on a cron from your infra. Want me to adapt this to: - Only delete logs, not the runs? - Keep runs but delete artifacts and logs? - A specific subset of flows or tags? If you tell me Cloud vs self-hosted and exactly what you want to retain, I’ll tailor the script and add a schedule.