Hi everyone, Can someone help me with adding dynamic tags for the task? Also, how to set different dynamic tags for different flow within a same script, or a subflow within a flow. I want to have different dynamic tags to each subflow. Thank you so much in advance for this help.

Hey Shahid! Here's an example for you that applies dynamic tags to subflows, the main flow, and tasks. LMK if it's close to what you're looking for. You don't have to use colors, of course, it can be whatever dynamic value you'd like. 🎨

from prefect import flow, get_client, task
from prefect.runtime import flow_run
from random import choice

#Update tags for flow runs
def update_flow_run_tags():
    COLORS = ["red", "blue", "green", "yellow", "purple", "orange", "pink"]
    #tags need to be a valid list
    random_color = [f"{choice(COLORS)}"]
    flow_run_id = flow_run.get_id()
    with get_client(sync_client=True) as client:
        client.update_flow_run(flow_run_id=flow_run_id, tags=random_color)

#Update tags for task runs``
def update_task_tags():
    COLORS = ["red", "blue", "green", "yellow", "purple", "orange", "pink"]
    random_color = [f"{choice(COLORS)}"]

    return random_color

@task(tags=update_task_tags())
def task1(name: str):
    print(f"Hello, {name}, I'm task 1!")
    
@task(tags=update_task_tags())
def task2(name: str):
    print(f"Hello, {name}, I'm task 2!")

@flow()
def subflow1(name: str):
    update_flow_run_tags()
    print(f"Hello, {name}, I'm subflow 1!")
    
@flow()
def subflow2(name: str):
    update_flow_run_tags()
    print(f"Hello, {name}, I'm subflow 2!")
    
@flow(log_prints=True)
def dynamic_tags_flow(name: str):
    #Update tags for the parent flow
    update_flow_run_tags()   
    print(f"Hello, {name}, I'm a main flow!")
    
    #Call the tasks
    task1(name)
    task2(name)
    
    #Call the subflows
    subflow1(name)
    subflow2(name)

fyi there's also a

tags

context manager designed for this:

from prefect import tags

@flow
def example():
    with tags("blue", "red", "green"):
        ... # any task called or submitted within this block
        ... # will inherit the new tags

^ good call, ty Chris!

Once again, thank you @Bianca Hoch and @Chris White for this help. I've tried this in the past and I tried it once again with the latest version, and it works. But it doesn't do what I'm expecting, in case of tags for TASK. When I apply this within a task, it tags the flow run it's associated with, and not the configuration of the task here (picture below). Let me know if my expectation is unreal.

That doesn't seem right - could you open an issue on github with a reproducible example for us?

Sure, I'll do that but in a day or two. Thanks for the quick reply.