<@ULVA73B9P> - is there a way with the API or CLI ...
# ask-marvin
k
@Marvin - is there a way with the API or CLI to find the artifacts created from a given flow (via flow run id)
m
thought for 170.8 seconds
Short answer: Yes via the API, not directly via the CLI. Assuming Prefect 3.x. Python API (all artifact versions for a flow run)
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterFlowRunId
from prefect.client.schemas.sorting import ArtifactSort

flow_run_id = UUID("00000000-0000-0000-0000-000000000000")

async def list_artifacts_for_run(flow_run_id: UUID):
    async with get_client() as client:
        artifacts = await client.read_artifacts(
            artifact_filter=ArtifactFilter(
                flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
            ),
            sort=ArtifactSort.UPDATED_DESC,
            limit=200,
        )
        for a in artifacts:
            print(a.id, a.key, a.type, a.created, a.updated)

# If you prefer sync:
# with get_client(sync_client=True) as client:
#     artifacts = client.read_artifacts(
#         artifact_filter=ArtifactFilter(
#             flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
#         ),
#         limit=200,
#     )
Python API (latest version per key only)
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    ArtifactCollectionFilter, ArtifactCollectionFilterFlowRunId
)

flow_run_id = UUID("00000000-0000-0000-0000-000000000000")

async def list_latest_artifacts_for_run(flow_run_id: UUID):
    async with get_client() as client:
        latest = await client.read_latest_artifacts(
            artifact_filter=ArtifactCollectionFilter(
                flow_run_id=ArtifactCollectionFilterFlowRunId(any_=[flow_run_id])
            ),
            limit=200,
        )
        for col in latest:
            # col.key is the artifact key; col.latest_id is the latest artifact version id
            print(col.key, col.latest_id)
Alternative filter (via FlowRunFilter)
Copy code
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId

async with get_client() as client:
    artifacts = await client.read_artifacts(
        flow_run_filter=FlowRunFilter(id=FlowRunFilterId(any_=[flow_run_id])),
        limit=200,
    )
REST API - Endpoint: POST /artifacts/filter (against your Prefect API) - Body:
Copy code
{
  "artifacts": {
    "flow_run_id": {
      "any_": [
        "00000000-0000-0000-0000-000000000000"
      ]
    }
  },
  "limit": 200
}
CLI - There isn’t a flag to filter by flow run id.
prefect artifact ls
lists artifacts, but has no
--flow-run-id
option:
Copy code
prefect artifact ls --help
You’ll need the Python client or REST API for run-scoped queries. If you’re on Prefect 2.x or need a ready-to-run helper script, let me know and I’ll tailor it.