Tom Han
09/03/2025, 5:25 PMMarvin
09/03/2025, 5:25 PMMarvin
09/03/2025, 5:29 PMflow_run_id
.
Here’s a pattern for option 2 using the Prefect client to target the parent run:
from prefect import flow
from prefect.artifacts import create_markdown_artifact
from prefect.runtime import flow_run
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.actions import ArtifactCreate
@flow
def child():
# Create an artifact for the child run (normal behavior)
md = "# Child summary\nThis is the child artifact."
create_markdown_artifact(md, key="child-summary")
# Also attach the same artifact to the parent flow run
parent_id = flow_run.parent_flow_run_id
if parent_id:
with SyncPrefectClient() as client:
client.create_artifact(
ArtifactCreate(
type="markdown",
key="child-summary", # optional, helps de-duplicate/update
description="Child summary published to parent run",
data={"markdown": md},
flow_run_id=parent_id, # <-- the key bit
)
)
@flow
def parent():
child()
if __name__ == "__main__":
parent()
Notes:
- In a subflow, prefect.runtime.flow_run.parent_flow_run_id
gives you the parent run ID; use that to target the parent.
- If you want to attach to the top-level run in deeper nesting, use prefect.runtime.flow_run.root_flow_run_id
.
- The convenience functions like create_markdown_artifact
attach to the current run; they don’t take a flow_run_id
. Use the client if you need to target another run.
- Use key
to manage updates/deduplication of artifacts across re-runs.
Docs:
- Artifacts overview: Artifacts in Prefect 3.x
- REST reference (create artifact): Create Artifact APINate
09/03/2025, 5:40 PM