Hello, Is there any way to dynamically name a task using a dictionary key as the task name? The below will fail.

@task(task_run_name="{input_dict['key']}")
def task(input_dict: dict):
    print(input_dict['key'])

Hey! I got this to work for me:

from prefect import task, flow

@task(task_run_name="{greeting}-{name}")
def my_task(name, greeting):
    print(f"I got a new name!")

@flow()
def my_flow():
    my_dictionary = {"name" : "Bianca"}
    return my_task(name= my_dictionary['name'], greeting = "My New Name is ")

my_flow()

That works, but creates an extra input to the task. In my use case, I’d be separating the dictionary out into separate inputs solely for the dynamic task name, which is fine, but not optimal.

Ah, understood. I'll see if I can reconfigure this so that the task isn't taking a extra argument

Hey, I had the same problem recently, here is a small utility I made to workaround it:

import pandas as pd


class TemplatedStr(str):
    """A string that can be formatted using items from a dict.

    Example:
    >>> d = {"user": {"name": "John"}, "work": {"tasks": [1, 2, 3]}}
    >>> s = TemplatedStr("Hello {user_name}, you have {len_work_tasks} tasks to do")
    >>> s.format(**d)
    "Hello John, you have 3 tasks to do"
    """

    def format(self, **kwargs):
        # convert {'a': {'b': 1}} into {'a_b': 1}
        _normalized = pd.json_normalize(kwargs, sep="_")
        _normalized_dict = _normalized.to_dict(orient="records")[0]
        # add len_* for all lists in normlized_dict
        _lenght_dict = {
            f"len_{k}": len(v)
            for k, v in _normalized_dict.items()
            if hasattr(v, "__len__")
        }
        parsed_kwargs = {
            **_normalized_dict,
            **_lenght_dict,
            **kwargs,
        }
        return super().format(**parsed_kwargs)