Hi <@ULVA73B9P>, I would like to add tags to tasks...
# marvin-ai
g
Hi @Marvin, I would like to add tags to tasks programatically, i.e. the tag is inferred at runtime. Hence I can't hard-code it in the decorator, and the curly bracket syntax that is supported for the task and flow run names seems not supported in the tags field. What are my other options?
m
thought for 131.5 seconds
Great question — you’re right: in Prefect 3.x, tags in decorators are static and the curly-brace templating you can use for run names isn’t supported for tags. Good news: you have two solid options for setting tags dynamically at runtime. 1) Use the tags context manager at the call site - Wrap the task submission (or call) in a
with tags(...):
block. Any task runs created in that block inherit those tags. - This is the most common way to drive tag-based concurrency limits or to scope tags to specific invocations. Example:
Copy code
from prefect import flow, task, tags

@task
def process(item):
    ...

@flow
def my_flow(customer_id: str, items: list[str]):
    runtime_tag = f"customer:{customer_id}"
    # Apply to all tasks in this block
    with tags(runtime_tag, "priority:high"):
        for item in items:
            process.submit(item)  # each run gets the dynamic tags

    # Or per-invocation if tags differ per item
    for item in items:
        with tags(f"item:{item}"):
            process.submit(item)
Notes: - Put the
with tags(...)
block in the flow (or the parent task) that creates the task run(s). Using it inside the body of a task won’t change that task’s own tags, but it will affect any nested tasks created within it. 2) Create a runtime-cloned task with with_options - You can create a “cloned” Task object at runtime with different tags, then call or submit it. Example:
Copy code
from prefect import flow, task

@task
def process(item):
    ...

@flow
def my_flow(customer_id: str, items: list[str]):
    runtime_tag = f"customer:{customer_id}"
    dynamic_process = process.with_options(tags=[runtime_tag, "priority:high"])

    for item in items:
        dynamic_process.submit(item)
This is useful if you want to set once and reuse within the flow run, but still compute the tags dynamically. When to use which: - Want tags to apply to a whole block of submissions? Use
with tags(...)
. - Want a single “dynamic default” task for the whole flow run? Use
with_options(...)
. A couple of reminders: - You can pass multiple tags:
with tags("a", "b")
or
with tags(*tag_list)
. - You cannot change tags after a task run has started. - If you’re using tags for concurrency limits, this pattern is exactly what you want. See: Tag-based concurrency limits in Prefect 3.x. If you can share how you’re using tags (e.g., concurrency limits, organization, filtering), I can suggest a pattern tailored to your use case.