Thread
#prefect-community
    k

    Kevin

    10 months ago
    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?
    I realize there are some easy solutions such as copying/pasting the flow and changing the name... but then whats the point of reusability?
    Anna Geller

    Anna Geller

    10 months ago
    @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.
    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

    10 months ago
    Awesome. This should work. Thanks
    does this only work with individual flow runs, not registered flows?
    Anna Geller

    Anna Geller

    10 months ago
    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:
    with Flow("your-flow-name") as flow:
    k

    Kevin

    10 months ago
    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