Samuel Kohlleffel

03/22/2023, 8:07 PM
Hello, Is there any way to dynamically name a task using a dictionary key as the task name? The below will fail.
def task(input_dict: dict):

Bianca Hoch

03/22/2023, 9:55 PM
Hey! I got this to work for me:
from prefect import task, flow

def my_task(name, greeting):
    print(f"I got a new name!")

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

would something like this suffice?

Samuel Kohlleffel

03/22/2023, 10:29 PM
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.

Bianca Hoch

03/23/2023, 5:23 PM
Ah, understood. I'll see if I can reconfigure this so that the task isn't taking a extra argument

Ouail Bendidi

03/29/2023, 9:05 AM
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.

    >>> 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 = {
        return super().format(**parsed_kwargs)
:gratitude-thank-you: 1
🚀 1