https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> is there a way I can use task or flow...
# ask-marvin
a
@Marvin is there a way I can use task or flow state values or other internal storage to store an internal "percentage complete" of my flow? e.g. I want to make that it's 10, 15,19,25% complete based on our criteria as it completes work and then retrieve this value from an external process
m
✅ thought for 139.5 seconds
Yes — in Prefect 3.x, the best fit for this is artifacts, specifically progress artifacts. Short answer: don’t use task/flow state values for arbitrary 
% complete
tracking. States are for orchestration status (
Running
, 
Completed
, 
Failed
, etc.), not for mutable custom progress data. If you want an external process to poll current progress while the flow is still running, use a progress artifact. Recommended pattern
Copy code
python
from prefect import flow, task
from prefect.artifacts import create_progress_artifact, update_progress_artifact
import time

@task
def do_work():
    artifact_id = create_progress_artifact(
        progress=0.0,
        key="my-flow-progress",
        description="Progress for my flow"
    )

    for pct in [10, 15, 19, 25, 50, 100]:
        # do some work
        time.sleep(1)

        update_progress_artifact(
            artifact_id=artifact_id,
            progress=pct
        )

@flow
def my_flow():
    do_work()

if __name__ == "__main__":
    my_flow()
Then an external process can retrieve the latest artifact by key through the Prefect API / client. Example with the Python client:
Copy code
python
import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterKey

async def read_progress():
    async with get_client() as client:
        artifacts = await client.read_artifacts(
            artifact_filter=ArtifactFilter(
                key=ArtifactFilterKey(any_=["my-flow-progress"])
            ),
            limit=1,
        )
        if artifacts:
            return artifacts[0].data
        return None

print(asyncio.run(read_progress()))
A few important notes: - This works during execution, not just after the flow finishes. - Progress artifacts are designed for values in the 
0-100
range. - They also show up nicely in the Prefect UI. - If you need more than a percentage — like 
items_processed
, 
current_step
, etc. — you may want to pair this with custom events or a separate external store. What not to use - Flow/task states: not intended for custom mutable progress - Task results: only really available once tasks finish - Logs: possible, but painful for external structured polling - Variables: technically possible, but not ideal for frequent updates If you want, I can also show you: 1. how to associate progress with a specific flow run instead of a global key, or 2. how to expose richer progress like 
{percent, step_name, completed_items}
. Relevant docs: - Artifacts - Progress artifacts - Events
a
@Marvin what is a progress artifact name? Can I attach it to a flow and query it like that or do I need to construct a specific name which includes my flow run name?
m
✅ thought for 104.0 seconds
A progress artifact has two relevant identifiers: - 
id
 — the unique UUID for that specific artifact record - 
key
 — an optional user-defined string you choose for lookup/grouping There isn't really a separate special "name" concept for progress artifacts beyond the 
key
. How this works in practice When you create one:
Copy code
python
artifact_id = create_progress_artifact(
    progress=0.0,
    key="my-progress"
)
- 
artifact_id
is what you use to update that same artifact in place - 
key="my-progress"
is what you use to find it later from another process Are artifacts attached to the flow automatically? Yes. If you create the artifact inside a running flow/task, Prefect automatically associates it with the current 
flow_run_id
(and 
task_run_id
if inside a task). So you do not need to manually embed the flow run name or flow run id into the key just to attach it to the run. Can you query "the artifact for this flow run"? Yes — you can query artifacts filtered by 
flow_run_id
. So there are two common patterns: 1. Static key + filter by flow run id Good if your external process already knows the 
flow_run_id
.
Copy code
python
key="progress"
Then query for artifacts where: - 
key == "progress"
- 
flow_run_id == <that run id>
This is usually the cleanest pattern. 2. Per-run key Good if your external process only knows a constructed key and you want direct lookup without additional filtering.
Copy code
python
key=f"progress-{flow_run_id}"
This works too, but it’s usually not necessary unless it simplifies your external consumer. Recommendation I’d usually suggest: - use a simple stable key like 
progress
- let Prefect auto-associate it to the run - query by 
flow_run_id
from the external process That avoids key explosion and keeps your code cleaner. Example:
Copy code
python
from prefect import flow
from prefect.artifacts import create_progress_artifact, update_progress_artifact
from prefect.runtime.flow_run import id as flow_run_id

@flow
def my_flow():
    artifact_id = create_progress_artifact(
        progress=0.0,
        key="progress",
        description="Flow progress"
    )

    update_progress_artifact(artifact_id, 10)
    update_progress_artifact(artifact_id, 25)
    update_progress_artifact(artifact_id, 100)
Then externally, query artifacts for: - 
flow_run_id=<run-id>
- optionally 
key="progress"
When should you include the run id in the key? Only if: - your external system cannot easily query by 
flow_run_id
- or you want a single direct lookup key per run - or you are not sure multiple runs may be active concurrently and your retrieval path only uses key lookup Otherwise, no, you do not need to include the flow run name/id in the key. One more subtle point: - If you create an artifact without a key, it is still attached to the run and visible on that run. - But if you want easy external lookup, give it a key. So for your use case, I’d likely do:
Copy code
python
key="percent-complete"
and then retrieve by run id + key. Docs: - Artifacts - Progress artifacts If you want, I can give you a complete example of: - flow code that updates progress, and - an external polling script that fetches 
percent-complete
for a specific running flow run.
Open in Slack
PreviousNext
Powered by