saatvik rao
12/30/2025, 11:04 AMMarvin
12/30/2025, 11:04 AMMarvin
12/30/2025, 11:06 AMflow_run_id using the Python client (or the REST API).
Python (async)
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)
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)
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?saatvik rao
12/30/2025, 11:11 AMMarvin
12/30/2025, 11:11 AMMarvin
12/30/2025, 11:15 AM.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:
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:
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.