Robert Szomolanyi
04/08/2026, 10:21 AMMarvin
04/08/2026, 10:21 AMMarvin
04/08/2026, 10:24 AMimport asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey
from prefect.client.schemas.sorting import ArtifactSort
async def delete_old_versions_for_key(key: str, keep_count: int = 1) -> dict:
async with get_client() as client:
# Get all versions for this key, newest first
artifacts = await client.read_artifacts(
limit=1000,
sort=ArtifactSort.CREATED_DESC,
artifact_filter=ArtifactFilter(key=ArtifactFilterKey(any_=[key])),
)
deleted = 0
kept = 0
for i, art in enumerate(artifacts):
if i < keep_count:
kept += 1
continue
await client.delete_artifact(artifact_id=art.id)
deleted += 1
return {"key": key, "kept": kept, "deleted": deleted, "total": len(artifacts)}
if __name__ == "__main__":
print(asyncio.run(delete_old_versions_for_key("my-artifact-key", keep_count=1)))
Notes
- This keeps the N most recent versions (default 1) and deletes the rest by ID.
- If you accidentally delete the latest, Prefect will automatically move the collection’s latest pointer to the next newest remaining version.
- For many keys, iterate your keys and call the function per key. You can get keys from latest collections via client.read_latest_artifacts(...).
CLI (manual) approach
- List all versions for a key (newest first):
prefect artifact inspect "my-artifact-key"
- Delete older versions by ID (repeat for each old ID):
prefect artifact delete --id <artifact_id>
- Important: prefect artifact delete <key> deletes all versions for that key (not what you want here).
Verified CLI help
- List: prefect artifact ls --help
- Inspect: prefect artifact inspect --help
- Delete: prefect artifact delete --help
Would you like this scripted to:
- keep the last N versions per key,
- or delete by age (e.g., older than X days) for a prefix/pattern of keys? I can tailor a complete snippet for your use case.Robert Szomolanyi
04/08/2026, 10:30 AMMarvin
04/08/2026, 10:30 AMMarvin
04/08/2026, 10:33 AMprefect.runtime.flow_run.ui_url. This works the same on a process worker, thread worker, etc.
Example:
from prefect import flow, get_run_logger
from prefect.runtime import flow_run
@flow
def my_flow():
logger = get_run_logger()
<http://logger.info|logger.info>(f"Flow run URL: {flow_run.ui_url}") # Full UI link to this run
return flow_run.ui_url
Notes:
- flow_run.ui_url returns a full link to the run in Prefect Cloud or your self-hosted UI, depending on your settings.
- If you need to build it manually, you can use:
from prefect.runtime import flow_run
from prefect.settings import PREFECT_UI_URL
url = f"{PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
- Outside a running flow/task context, flow_run.ui_url will be None.