<@ULVA73B9P> How to delete old versions of artifac...
# ask-marvin
r
@Marvin How to delete old versions of artifact except newest one ?
m
thought for 159.7 seconds
Short answer: there isn’t a one-liner to “delete all but the newest” in Prefect. In Prefect 3.x you can do it either: - programmatically with the Prefect client (recommended), or - manually via the CLI by deleting specific artifact IDs. Programmatic cleanup (keep only the newest version for a given key)
Copy code
import 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):
Copy code
prefect artifact inspect "my-artifact-key"
- Delete older versions by ID (repeat for each old ID):
Copy code
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.
r
@Marvin Is it possible to get UI server URL from flow, which is running in process worker ?
m
thought for 127.6 seconds
Yes. In Prefect 3.x you can get the UI URL for the current flow run from inside your flow or task via
prefect.runtime.flow_run.ui_url
. This works the same on a process worker, thread worker, etc. Example:
Copy code
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:
Copy code
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.