<@ULVA73B9P> is that possible to create an artifac...
# ask-marvin
s
@Marvin is that possible to create an artifact outside a task context and attach it to a specific task of flow run ?
m
thought for 421.2 seconds
Short answer: yes. In Prefect 3.x you can create an artifact outside of any task/flow context and explicitly attach it to a specific task run by setting the artifact’s task_run_id (and optionally flow_run_id). The convenience helpers like
create_markdown_artifact(...)
do not accept run IDs; use the artifact classes directly and call `.create()`/`.acreate()`. Example (async) that looks up a task run by name within a flow run, then creates a Markdown artifact attached to that task run:
Copy code
from uuid import UUID
from prefect.client.orchestration import PrefectClient
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId, TaskRunFilterName
from prefect.artifacts import MarkdownArtifact

async def attach_markdown_to_task(flow_run_id: UUID, task_name: str, markdown: str, key: str | None = None):
    async with PrefectClient() as client:
        task_runs = await client.read_task_runs(
            task_run_filter=TaskRunFilter(
                flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id]),
                name=TaskRunFilterName(any_=[task_name]),
            )
        )
        if not task_runs:
            raise RuntimeError(f"No task runs named {task_name!r} found in flow run {flow_run_id}")

        task_run_id = task_runs[0].id

        artifact = MarkdownArtifact(
            markdown=markdown,
            key=key,  # optional; if provided, must be lowercase letters, numbers, and dashes
            task_run_id=task_run_id,
            flow_run_id=flow_run_id,
        )
        artifact_id = await artifact.acreate()
        return artifact_id
Notes: - If you are already inside the target task run (e.g., inside the function executed via
.delay()
on a task worker), just call the helpers like
create_markdown_artifact(...)
— Prefect will infer the run IDs from context. - Creating artifacts entirely “unattached” (no run context and no IDs) is discouraged; you’ll see a warning. If you must create them from your backend, pass the
task_run_id
(and
flow_run_id
) as shown above. - Since you dynamically name tasks with
.with_options(name=...)
, filtering by
TaskRunFilterName
will match that name. Docs: - Artifacts guide - prefect.artifacts Python API - Filters API - Background tasks If you’d like, I can tailor this to your exact pattern (e.g., your backend initiating artifacts after
asyncio.gather
) or provide a sync version.