https://prefect.io
Join Slack
still struggling in Prefect 2 to dynamically name mapped tasks like in Prefect 1: ```# callable name...
# ask-community
d
still struggling in Prefect 2 to dynamically name mapped tasks like in Prefect 1:
Copy code
# callable name generator for mapped task
def generate_task_run_name_foo(dataset: dict, **kwargs):
    return 'Foo:  ' + dataset['name']


# task that gets mapped in the flow
@task(... , task_run_name=generate_task_run_name_foo)
def foo(dataset: dict, ...):
	...
Prefect 2 doesn’t seem to pass the args of foo into the name generator like v1 does. Is there a different way of achieving this in v2? Prefect 2 transcription leads to 
missing 1 required positional argument: 'dataset'
✅ 1
d
have you tried 
with_options
method of Task?
d
same error with that vs. using the task decorator
d
Do you want a random name generated for the task by Python OR do you want to implement a function that takes in input and generates task name based on the inputs?
d
task name based on inputs… specifically in the case of mapped tasks
d
ah mapped tasks are tricky - not sure about them sorry
d
😕 no problem. The code block above works for all my Prefect 1 flows but it just doesn’t seem 1-to-1 while migrating
j
Hey! https://docs.prefect.io/2.10.12/concepts/tasks/#task-arguments under this section there is a few examples that I think cover what you need
One of the examples:
Copy code
from prefect import flow, task
from prefect.runtime import flow_run, task_run

def generate_task_name():
    flow_name = flow_run.flow_name
    task_name = task_run.task_name

    parameters = task_run.parameters
    name = parameters["name"]
    limit = parameters["limit"]

    return f"{flow_name}-{task_name}-with-{name}-and-{limit}"

@task(name="my-example-task",
      description="An example task for a tutorial.",
      task_run_name=generate_task_name)
def my_task(name: str, limit: int = 100):
    pass

@flow
def my_flow(name: str):
    # creates a run with a name like "my-flow-my-example-task-with-marvin-and-100"
    my_task(name="marvin")
d
Thank you @Jake Kaplan! This worked. i switched the callable name generator in my original code block to this:
Copy code
def generate_task_run_name_foo() -> str:
    return 'Foo:  ' + task_run.parameters['dataset']['name']
🙌 1
t
Hi @Jake Kaplan I'm going through the above as I'd also like to rename mapped tasks and get the error:
Copy code
AttributeError: 'function' object has no attribute 'format'
from the setup:
Copy code
from prefect import task
from prefect.runtime import task_run

def generate_archive_task_name() -> str:
    schema_name = task_run.parameters["schema_name"]
    table_name = task_run.parameters["table_name"]
    return f"archive__{schema_name}__{table_name}"

@task(task_run_name=generate_archive_task_name)
def archive_table(schema_name: str, table_name: str) -> None:
    ...
currently on prefect 2.10.5 - was this implemented more recently?
j
Unfortunately it was added in 2.10.6 https://github.com/PrefectHQ/prefect/pull/8933
👍 1
t
thanks so much!
6 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by