@Marvin I have the following task to delete old prefect runs @task async def _prune_old_prefect_flows_(_days_: int = 56, _batch_size_: int = 200): """ Remove old flow logs to avoid massive disk usage Args: days (int, optional): how many days old flows should be deleted. Defaults to 56. batch_size (int, optional): Stops overwhelming system. Defaults to 200. """ log = _get_run_logger_() cutoff = datetime._now_(timezone.utc) - _timedelta_(days=days) deleted = 0 # Prefect 3 deletes flow-run logs after flow-run deletion. async with _get_client_() as client: while True: runs = await client._read_flow_runs_( _flow_run_filter_=_FlowRunFilter_( _start_time_=_FlowRunFilterStartTime_(before_=cutoff), state=_FlowRunFilterState_( type=_FlowRunFilterStateType_( any_=[ StateType.COMPLETED, StateType.FAILED, StateType.CANCELLED, StateType.CRASHED, ] ) ), ), limit=_batch_size_, ) if not runs: break for run in runs: await client._delete_flow_run_(_flow_run_id_=run.id) deleted += 1 log._info_(_f_"Deleted flow run: {run.name} ({run.id})") log._info_(_f_"Deleted {deleted} flow runs so far (cutoff: {cutoff._isoformat_()})") log._info_(_f_"Done. Deleted {deleted} flow runs older than {cutoff._isoformat_()}") and call it in a flow: @flow def _weekly_maintenance_(_retention_days_: int = 14): log = _get_run_logger_() log._info_(_f_"Starting cleanup — retention: {_retention_days_} days") _cleanup_workspace_logs_(_retention_days_) _cleanup_job_logging_storage_(_retention_days_) _vacuum_airbyte_db_() _truncate_container_logs_(_max_size_mb_=100) _prune_docker_images_() asyncio._run_(_prune_old_prefect_flows_()) _restart_airbyte_() log._info_("Cleanup complete") firstly is my clearing of flows good practice to avoid storage build up, secondly is the task going to work without errors and without pitfalls, and thirdly, since it has to be an async task, can i call it in my flow with asyncio and it will work

✅ thought for 221.2 seconds

hey @Pav Staz - you may be interested in the new service that will be released on thursday https://github.com/PrefectHQ/prefect/pull/20746

Great questions — a retention/cleanup flow like this is a good idea, and you’re very close. Here’s a concise review and the recommended fixes. 1) Is deleting old flow runs good practice? - Yes, especially for self-hosted Prefect Server where the database (and logs) can grow indefinitely. Deleting older, terminal-state runs keeps your DB lean and your UI snappy. - When a flow run is deleted in Prefect 3, its logs are removed by a background task shortly after. So you don’t need to delete logs separately. - Caveat: filtering by start_time won’t catch runs that never started (no start_time). That’s a known limitation; it’s generally acceptable but worth noting. References: - Flow-run deletion endpoint (and background log cleanup): server/api/flow_runs.py - Client methods and filters you’ll use: client/orchestration/_flow_runs/client.py, client/schemas/filters.py 2) Will your task work without errors and pitfalls? - With a few syntax fixes, yes. Specifically: - Use

client.read_flow_runs(...)

and

client.delete_flow_run(...)

(no leading/trailing underscores). - Use filter field names like

before_

and

any_

(single trailing underscore), not `before__`/`any__`. - Include a short sleep between batches to be kind to your DB/API. - Stick to terminal states (COMPLETED/FAILED/CANCELLED/CRASHED) to avoid deleting active runs. - Note: logs are deleted asynchronously after the run is deleted, so space won’t free up exactly at the same moment. - Optional: if you want absolute stability, you can add simple retry handling around deletes, but it’s usually not necessary. Here’s a corrected, drop-in version:

from datetime import datetime, timedelta, timezone
import asyncio

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


@task
async def prune_old_prefect_flows(days: int = 56, batch_size: int = 200):
    """
    Delete terminal flow runs older than `days`.
    Prefect 3 will remove associated flow-run logs shortly after via a background task.
    """
    log = get_run_logger()
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    deleted = 0

    run_filter = FlowRunFilter(
        start_time=FlowRunFilterStartTime(before_=cutoff),
        state=FlowRunFilterState(
            type=FlowRunFilterStateType(
                any_=[
                    StateType.COMPLETED,
                    StateType.FAILED,
                    StateType.CANCELLED,
                    StateType.CRASHED,
                ]
            )
        ),
    )

    async with get_client() as client:
        while True:
            runs = await client.read_flow_runs(
                flow_run_filter=run_filter,
                limit=batch_size,
            )
            if not runs:
                break

            for run in runs:
                await client.delete_flow_run(flow_run_id=run.id)
                deleted += 1
                <http://log.info|log.info>(f"Deleted flow run: {run.name} ({run.id})")

            <http://log.info|log.info>(f"Deleted {deleted} flow runs so far (cutoff: {cutoff.isoformat()})")
            # Small pause between batches to reduce DB/API pressure
            await asyncio.sleep(1.0)

    <http://log.info|log.info>(f"Done. Deleted {deleted} flow runs older than {cutoff.isoformat()}")

3) How should you call an async task from your flow? - Don’t call

asyncio.run()

inside a flow — Prefect manages the event loop and this can raise “asyncio.run() cannot be called from a running event loop”. - Instead:

@Nate i assume no docs for that. I assume I just install latest version of prefect? How will I set the retention period?

we'll add docs, just making you aware of the recent contribution. it will be released in 3.6.19 on thursday, we'll have docs at that point