The flow run name can not be set in advance, but it can be changed using the
RenameFlowRun task after the flow run has been created.
You can use this task inside the Flow block or through a flow-level state handler. When calling the task from a state handler, make sure to call the task’s
Here is a flow-level state handler example:
from prefect import Flow, task
from prefect.tasks.prefect import RenameFlowRun
def rename_handler(flow, new_state, old_state):
with Flow("test-flow", state_handlers=[rename_handler]) as flow:
There is incorrect order of states in your example.
Handler should have following signature:
state_handler(flow: Flow, old_state: State, new_state: State) -> Optional[State]
@Nikita_Samoylov! It actually doesn’t matter because regardless of whether the
obj is a
Task object, it will be passed to the state handler, depending on whether you attach the state handler to a task or a flow. But I agree with you that renaming it to flow makes it more explicitly, so I’ll change it.
Thanks for posting this!
October 18, 2022, 3:28am
Hi, I got an error: Import “prefect.tasks.prefect” could not be resolved
I’m using Prefect 2.4.5
this is Prefect 1 syntax, don’t use it in Prefect 2
Docs for Prefect 2 are available here:
How would you do this in Prefect 2? I would like to overwrite the randomly generated combinations but there doesnt seem to be a
RenameFlowRun equivalent in 2.0?
we have an open feature request for it here:
06:10PM - 07 Nov 22 UTC
### First check
- [X] I added a descriptive title to this issue.
- [X] I used t
… he GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.
### Prefect Version
### Describe the current behavior
Tasks & flow runs get random names, which makes it hard to track correct behavior, and identify problem sources. Although the metadata of those runs contain the required information (e.g. name of processed item, time of execution, sequence number), those should be checked out separately.
### Describe the proposed behavior
With Prefect v1 one could use `RenameFlowRun` and `task_run_name` to rename a flow or task run by a template. Prefect v2 provides `update_flow_run` for the former, but it's usage is quite complicated for this basic task. Also, for tasks `.with_options(name=...)` could not be applied when mapping is used.
I propose to have `task_run_name` again for tasks, and `flow_run_name` for flows, which would define a naming template for the respective entities.
### Example Use
Processing collections of items could be a frequent use case. Usually each item has an own name or identifier. For example a directory of files, a list of queries, a batch of urls, a set of clients read from a database, where the list of items to be processed is not known in advance. A natural way of processing these would be to use mapping or a loop, but random (cool)names doesn't help to know what's going on, but a `task_run_name` template could.
Flow runs may be bound to a period of time (e.g. Monday, August, 45th week), or a sequence number (1,2,3,4,5...). If there was a problem, and those values could show up in the name, then it would be immediately obvious what was affected. A `flow_run_name` parameter of the flow containing a naming template could help with this.
### Additional context
and here is a workaround solution until then contributed by another user:
This file has been truncated.
from prefect import flow, task, get_run_logger
from random import randint, random
def rename_flow_run(new_name: str):