<@ULVA73B9P> can I add tags flow run, while a flow...
# ask-community
j
@Marvin can I add tags flow run, while a flow or task is executing
m
thought for 354.6 seconds
Short answer: - Flow runs: yes — you can update a flow run’s tags while it is executing. - Task runs: no — once a task run starts, its tags are fixed; you can only tag future task runs that haven’t started yet. Details and examples Flow runs (can update mid-run) - Use the orchestration client to update tags on the current flow run. Note that updates replace the entire list, so merge with existing tags if you want to “add” rather than overwrite. Example:
Copy code
from prefect import flow, get_client
from prefect.runtime import flow_run

@flow
async def my_flow():
    async with get_client() as client:
        # Read existing tags
        fr = await client.read_flow_run(flow_run.id)
        current = set(fr.tags or [])
        # Add a tag and update
        new_tags = sorted(current | {"hotfix"})
        await client.update_flow_run(flow_run.id, tags=new_tags)
- Read the current flow run’s tags at runtime:
Copy code
from prefect.runtime import flow_run
print(flow_run.tags)  # list[str] or None
Task runs (cannot update mid-run) - There is no API to mutate a task run’s tags after it has started. - To influence tags during a flow’s execution, apply tags to future task runs using: - The
@task(tags=[...])
decorator (static) - The
with tags(...):
context manager around the task call (dynamic at call time, applies to runs created inside the context) Example for tagging future task runs:
Copy code
from prefect import flow, task, tags

@task
def do_work(x): 
    return x

@flow
def tagged_tasks():
    # This run gets the "priority" tag
    with tags("priority"):
        do_work(1)
    # This run gets the "low" tag
    with tags("low"):
        do_work(2)
Good to know - Tag-based concurrency decisions are made when a run is created. Updating flow run tags mid-execution won’t retroactively change concurrency decisions already made for existing task runs, but it can impact automations or any logic that inspects flow run tags later. - Accessing the current flow run id is a property, not a function call:
flow_run.id
. References - Tags context manager: https://docs.prefect.io/v3/api-ref/python/prefect-context/#tags - Runtime flow_run values (e.g., id, tags): https://docs.prefect.io/v3/api-ref/python/prefect-runtime-flow_run/ - Client API (update_flow_run): https://reference.prefect.io/prefect/client/orchestration/__init__/#prefect.client.orchestration.PrefectClient.update_flow_run - Tag-based concurrency limits: https://docs.prefect.io/v3/how-to-guides/workflows/tag-based-concurrency-limits/ If you can share your use case (e.g., adding a tag to influence a running automation vs. controlling task-level concurrency), I can suggest the best pattern.