<@ULVA73B9P> how do i fetch artifacts of a flow us...
# ask-marvin
s
@Marvin how do i fetch artifacts of a flow using flow's run id?
m
thought for 168.6 seconds
In Prefect 3.x, fetch all artifacts for a specific flow run by filtering on
flow_run_id
using the Python client (or the REST API). Python (async)
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterFlowRunId

async def fetch_artifacts(flow_run_id: str | UUID):
    flow_run_id = UUID(flow_run_id) if isinstance(flow_run_id, str) else flow_run_id
    async with get_client() as client:
        artifacts = await client.read_artifacts(
            artifact_filter=ArtifactFilter(
                flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
            )
        )
    return artifacts
Python (sync)
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterFlowRunId

def fetch_artifacts_sync(flow_run_id: str | UUID):
    flow_run_id = UUID(flow_run_id) if isinstance(flow_run_id, str) else flow_run_id
    client = get_client(sync_client=True)
    return client.read_artifacts(
        artifact_filter=ArtifactFilter(
            flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
        )
    )
REST API (useful outside Python)
Copy code
curl -X POST "$PREFECT_API_URL/api/artifacts/filter" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -d '{
    "flow_runs": { "id": { "any_": ["<FLOW_RUN_ID>"] } },
    "limit": 200
  }'
Optional filters you can add: - By type (e.g., only tables):
ArtifactFilterType(any_=["table"])
- By key pattern:
ArtifactFilterKey(like_="report-%")
Want this via Python or curl in your environment?
s
@Marvin how to access data parameter in all the fetched artifacts?
m
thought for 213.1 seconds
Great question. In Prefect 3.x, each artifact has a
.data
field, but its type depends on the artifact type. The only special case is tables: table artifacts store their data as a JSON string, so you need to
json.loads()
them first. Everything else is already a native type. Here’s a safe pattern to access
.data
for all fetched artifacts:
Copy code
import json

def extract_artifact_data(artifacts):
    results = []
    for a in artifacts:
        if a.data is None:
            results.append(None)
            continue

        if a.type == "table" and isinstance(a.data, str):
            # Table data is JSON-encoded; parse it
            try:
                results.append(json.loads(a.data))
            except json.JSONDecodeError:
                # Fallback if somehow not JSON
                results.append(a.data)
        else:
            # Other types are already native:
            # - markdown/link: str
            # - image: str (URL)
            # - progress: float
            # - custom: could be str/dict/primitive
            results.append(a.data)
    return results
End-to-end example, continuing from your flow-run filtered fetch:
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterFlowRunId
import json

async def fetch_and_access_data(flow_run_id: str | UUID):
    flow_run_id = UUID(flow_run_id) if isinstance(flow_run_id, str) else flow_run_id
    async with get_client() as client:
        artifacts = await client.read_artifacts(
            artifact_filter=ArtifactFilter(
                flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
            )
        )

    data_values = []
    for a in artifacts:
        if a.data is None:
            data_values.append((a.key, a.type, None))
            continue

        if a.type == "table" and isinstance(a.data, str):
            value = json.loads(a.data)
        else:
            value = a.data

        data_values.append((a.key, a.type, value))

    return data_values  # list of tuples: (key, type, data)
Quick reference for
.data
by type: - table: JSON string → use
json.loads(artifact.data)
- markdown/link: string - image: string (URL) - progress: float (0–100) - custom: could be dict/str/primitive; handle accordingly If you share how you plan to use the data (e.g., convert tables to pandas, extract links, etc.), I can tailor the snippet.