When creating re-usable flows, how do you name them uniquely? For example, if I have a flow that connects to an SFTP, Downloads those files, and then writes them to cloud storage based off passed parameters - how does the flow name align with the parameters being passed?
Kevin
11/23/2021, 8:29 PM
I realize there are some easy solutions such as copying/pasting the flow and changing the name... but then whats the point of reusability?
a
Anna Geller
11/23/2021, 8:36 PM
@Kevin you can use a state handler to set a custom flow run name based on Parameter values. Currently, you can only do it by renaming an automatically assigned flow run name, but it will have the effect you want.
Copy code
from prefect import Flow, task
from prefect.tasks.prefect import RenameFlowRun
import prefect
def rename_handler(obj, new_state, old_state):
if new_state.is_running():
param = prefect.context.parameters.get("your_parameter_name")
RenameFlowRun().run(flow_run_name=f"new_name_{param}")
return
@task
def first_task():
return 1
with Flow("test-flow", state_handlers=[rename_handler]) as flow:
first_task()
k
Kevin
11/23/2021, 8:37 PM
Awesome. This should work. Thanks
Kevin
11/23/2021, 8:40 PM
does this only work with individual flow runs, not registered flows?
a
Anna Geller
11/23/2021, 8:42 PM
Flow run names are assigned at runtime, you can then rename those names after they have been created. To set a name for a flow, you need to set it on the Flow constructor:
Copy code
with Flow("your-flow-name") as flow:
k
Kevin
11/23/2021, 8:47 PM
okay i got it. was making it too difficult. register the flow as its generic-self and then create flow runs with unique names if desired
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.