Hi All I think I am missing something obvious but when I re-register a flow after making a minor change, the changes do not get picked up.

Hi @CM, how are you registering?

If I run something like this: flow.executor = prefect.executors.LocalExecutor() _flow.register(project_name='test project')_ Then kick it off in the UI everything goes as expected. However -- continuing with the same flow object -- after I run: _flow.executor = prefect.executors.LocalDaskExecutor(scheduler="processes", num_workers=16)_ _flow.register(project_name='test project')_ It gives me a UserWarning: A flow with the same name is already contained in storage; if you changed your Flow since the last build, you might experience unexpected issues and should re-create your storage object. Then if kick it off in the UI, it still runs under LocalExecutor()

No that’s not I think. I think this is because the Executor and Result are not part of the metadata we store. If you print

flow.serialize()

, you’ll find the executor is not there. This is because executor and results can contain sensitive information so we don’t keep them in the flow schema.

I've tried that too but with no luck. Let me try again. Also, executor does get stored somewhere associated with a flow? Since kicking off the flow from UI does honor the specified executor (assuming it gets registered correctly)

Yes with the Flow

ok, that makes sense. and since the flow is not registering properly, it is not being honored.

Does your

prefect register flow

command have a

force

option on that version?

I don't see that option when I run prefect register flow --help There is however a --skip-if-flow-metadata-unchanged option

I downgraded to replicate and yeah there’s no force option. What do you mean by re-generating?

just create another independent flow object identical to the one I want to modify

How do you register? With the CLI? Because when I use the

flow.register()

and then

python myflow.py

to register, I get the version bumped.

I also get a version bumped and the ID in the UI matches what I get from the flow.register() call But that's not what gets run -- hence my confusion

Ahh I see… ok so if you can upgrade, there is the

--force

version in the future. If you can’t, I would honestly try altering the Flow a little bit, maybe adding a log statement somewhere to force the metadata to change and I think that would force the re-registration.

yep, that makes sense. I will upgrade but in my use case I do not plan to use CLI for registering flows. I will try logging (I tried labels but that did not do it)

I tested it in 0.14.21 and changing the executor with

python myflow.py

does take effect. I changed from LocalDaskExecutor to LocalExecutor. You can run the flow with DEBUG logs to see the executor type.

If I understand what you mean, when you run python myflow.py and then change executor and run python myflow.py again, that's equivalent to me generating a second flow object and changing the executor, right? with Flow('my_flow') as flow: run_my_task() flow.executor = prefect.executors.LocalDaskExecutor(scheduler="processes", num_workers=16) flow.register('my_project') with Flow('my_flow') as flow: run_my_task() flow.executor = prefect.executors.LocalExecutor() flow.register('my_project') which does work for me as well.

Yes I think so

one last question: any way to see on the UI what was the flow ID that was run for a given Flow Run?

Deleted my messages as I understood what you meant.

yep, every version of a flow I register under the same name gets its own flow id. but when I look at flow run in the UI, how do I know which version (flow id) was run?

I guess there’s no other way but to use the graphQL API with

query {
  flow_run(where: { id: {_eq: "160d2a2b-58e1-442a-818d-2855f283ac6d"} }) {
    id
    flow_id
    flow {
      name
      version
    }
  }
}

Sounds good. Thanks!