Let me show you how to dynamically change task names in a loop with a practical example. There are two main approaches you can use: 1. Using a dynamic

task_run_name

with a function:

python
from prefect import flow, task

@task(task_run_name=lambda context: f"process-item-{context.parameters['item_number']}")
def process_item(item_number: int):
    return f"Processed item {item_number}"

@flow
def main_flow():
    results = []
    for i in range(3):
        result = process_item(item_number=i)
        results.append(result)
    return results

2. Using task mapping (more efficient for parallel execution):

python
from prefect import flow, task

@task(task_run_name=lambda context: f"process-item-{context.parameters['item_number']}")
def process_item(item_number: int):
    return f"Processed item {item_number}"

@flow
def main_flow():
    # This will create tasks named "process-item-0", "process-item-1", "process-item-2"
    results = process_item.map(item_number=[0, 1, 2])
    return results

Key points to understand: 1. The

task_run_name

parameter accepts either a string or a callable 2. When using a callable, it receives a context object that contains: - `parameters`: The parameters passed to the task - Other context information like run_id, task_run_id, etc. 3. You can also use a more complex function for naming:

python
def generate_task_name(context):
    item_num = context.parameters['item_number']
    return f"process-item-{item_num}-{context.flow_run.name}"

@task(task_run_name=generate_task_name)
def process_item(item_number: int):
    return f"Processed item {item_number}"

The second approach using

.map()

is generally recommended when you're processing multiple items, as it: - Handles parallelization automatically - Is more efficient than manual loops - Provides better visibility in the Prefect UI - Maintains unique task names for each iteration Choose the first approach (manual loop) when you need more control over the execution flow or when tasks need to be executed sequentially. Use the second approach (mapping) when you want parallel execution and have independent items to process. Each task run will appear in the Prefect UI with its unique name, making it easier to track and debug your workflow executions.

@Marvin i get the following error : TypeError: <lambda>() missing 1 required positional argument: 'context'

Based on the research and GitHub issues, I can confirm that the corrected version I provided above is the right approach. The

task_run_name

callable receives a dictionary of the task's parameters directly. A few additional tips: 1. Make sure your parameter names don't conflict with Prefect's internal context names 2. If you need access to more context information beyond the parameters, you might need to use a different approach (like task tags or states) 3. When using

.map()

, each mapped task will get its own unique name based on its parameters Try the corrected version and let me know if you run into any other issues!