@Marvin can I dynamically set a task run name if I'm using

mytask.map(inputs)

✅ thought for 493.4 seconds

Yes. https://docs.prefect.io/v3/how-to-guides/workflows/write-and-run#task-configuration Use the

task_run_name

parameter on the

@task

decorator. It accepts a string or method name.

@task(name="My Example Task", 
      description="An example task for a tutorial.",
      task_run_name="hello-{name}-on-{date:%A}")
def my_task(name, date):
    pass

Yes. In Prefect 3.x you can set a dynamic task run name for each mapped item by giving the task a

task_run_name

callable. The callable receives the bound parameters for that specific invocation, so it works perfectly with

.map(...)

. Example 1 — name based on a single mapped parameter:

from prefect import flow, task

@task(task_run_name=lambda p: f"upper-{p['text']}")
def upper(text: str) -> str:
    return text.upper()

@flow
def main():
    texts = ["alpha", "beta", "gamma"]
    futures = upper.map(texts)  # names: upper-alpha, upper-beta, upper-gamma
    return futures

Example 2 — include an index by mapping an index alongside your data:

from prefect import flow, task

@task(task_run_name=lambda p: f"process-{p['i']}-{p['item']}")
def process(item: str, i: int) -> str:
    return f"{i}:{item}"

@flow
def main():
    items = ["A", "B", "C"]
    futures = process.map(items, range(len(items)))
    return futures

Alternative if you’re not using `.map`: create a per-call variant with a fixed string via

.with_options(...)

and `submit`:

for i, item in enumerate(items):
    process.with_options(task_run_name=f"process-{i}-{item}").submit(item=item, i=i)

References - Task decorator/options (shows

task_run_name

and

with_options

): prefect/src/prefect/tasks.py -

Task.map

signature: prefect/src/prefect/tasks.py If you’re on Prefect 2.x instead, let me know — the approach is similar but details differ.

@Brendan Dalpe what if I want to set it inside the task? Basically I have a list of configs with are the inputs that get passed to .map() but then i want to dynamically set the task name based on a key inside the config it was passed that way I know which task is doing what exactly

I was wondering if I could use prefect.runtime. and update task_run.name

No, the

prefect.runtime.task_run

context is read-only.

what if I want to set it inside the task?

It's not possible to dynamically change the name inside the task method. You could dynamically set the name before you call the method:

@task(name="My task")
def my_task():
    return 1

@flow
my_flow():
    new_task = my_task.with_options(name="My new task")
    new_task()