Set name for flow run and (mapped) task runs

I am just starting to get acquainted with Prefect 2, but I could not really track what is happening, because coolnames are not very informative, just a bunch of random strings. I prefer to have names including a characteristic parameter (e.g. name of processed item) and a sequence number and/or some kind of date-time, whichever makes it unique.

Prefect 1.x has RenameFlowRun and task run names could be templated with task_run_name. Is there anything in Prefect 2.x that can be used to achieve the same easily?

Most of the time, I have to process collection of items (queries, files, urls, etc.), which are dynamically generated. I guess mapping was made for these, but it seems that there is no way to differentiate among the items in their naming. Is it so, or am I using it wrong?

This issue was closed because technically this is already possible:

if this doesn’t satisfy your use case, do you want to open a feature request on GitHub?

Thanks for your reply Anna!

It seems then, that there’s no easy way. I am going to open a feature request as you’ve suggested.

Also, I’ve tried to use update_flow_run, but no use. It looks like that it does not affect the running flow.

In the code below, set_flowrun_name task should rename the flow run to “whatever”, isn’t it?

from prefect import flow, task, context
from prefect.orion.schemas.actions import FlowRunUpdate
from prefect.orion.api.flow_runs import update_flow_run
from pprint import pprint

@task
def get_names():
    return ["Amber", "Barbara", "Charlie"]

@task(tags=['max_two_tasks'])
def say_hello(name):
    print(f"Hello {name}!")

@task
def dump_ctx():
    ctx = context.get_run_context()
    pprint(ctx)

@task
def set_flowrun_name():
    ctx = context.get_run_context()
    update_flow_run(FlowRunUpdate(name='whatever'), ctx.task_run.flow_run_id)

@flow(name="hello_name")
def hello_flow():
    flow_run_name = set_flowrun_name()
    names = get_names()
    hellos = [say_hello.with_options(name=f'hello-{name}')(name) for name in names]
    dump_ctx()

if __name__ == '__main__':
    hello_flow()
1 Like

I’m not sure whether it’s that easy to do it client-side - I suggest opening an issue and putting all the relevant information there, and then linking the issue here for posterity to follow up

Thanks so much!

Thank you for your code andor.

Here is the solution:

from prefect.orion.database.dependencies import provide_database_interface
@task
def set_flowrun_name():
    ctx = context.get_run_context()
    update_flow_run(FlowRunUpdate(name='whatever'), ctx.task_run.flow_run_id, provide_database_interface())

Great! Thanks @ssilent!

For reference, I have sent a feature request about this, which was added to the roadmap.

1 Like