Hello I m trying to register flows that are generated dynami

Question

Hello I m trying to register flows that are generated dynamically based on a yaml file and that are stored in a list In short I m using code like this flows is a simple python list of Flow objects ```for flow in flows flow register project name=f Data Platform os environ get PREFECT TARGET ENV Develop add default labels=False set schedule active=True if os environ get PREFECT TARGET ENV Develop == Production else False ``` This has worked for me before but since I upgraded prefect to 1 2 1 I m getting an error message related to the Azure storage I m using for the flows ```azure core exceptions ResourceExistsError The specified blob already exists``` Any idea why this strategy is not working anymore

Kevin Kho · Answer

This looks like you use Azure storage and the first registration already uploaded a file at that path and the second one will not overwrite it

Bob De Schutter · Answer

but why is it uploading the same file twice the different generated flows have different names

Bob De Schutter · Answer

``` 2022 05 20 16 54 23+0200 INFO prefect Azure | Uploading dbt datawarehouse daily 2022 05 20t14 54 22 329817 00 00 to datawarehouse prefect flows Flow URL <https cloud prefect io gv energy group flow 4c3ff3d7 ac87 43f4 bfe8 ded70fb20b22> ID c759911b fb92 493e 90e4 1925f0663d44 Project Data Platform Develop Labels 2022 05 20 16 54 25+0200 INFO prefect Azure | Uploading dbt datawarehouse daily 2022 05 20t14 54 22 329817 00 00 to datawarehouse prefect flows Traceback most recent call last ```

Bob De Schutter · Answer

so the first flow that gets generated is successfully registered and uploaded but then it seems that the second flow in the list is uploaded using the same name for some reason while the name is different slightly smiling face

Kevin Kho · Answer

How did you define storage

Bob De Schutter · Answer

```STORAGE=Azure container= datawarehouse prefect flows connection string secret= azure container connection string ```

Bob De Schutter · Answer

and the flows are created like this in a loop ```with Flow name=f dbt flow name storage=STORAGE run config=DockerRun image=DOCKER IMAGE schedule=schedule as dbt flow ```

Bob De Schutter · Answer

strange thing is that this exact same code was working before

Kevin Kho · Answer

I get what you are saying This is hard to say though I think the attempt is right I would just try adding more log statements in that loop to see the flow

Bob De Schutter · Answer

I quickly added a print statement in the loop registering the flows just before the register ``` lt Flow name= dbt datawarehouse daily gt 2022 05 20 17 08 01+0200 INFO prefect Azure | Uploading dbt datawarehouse daily 2022 05 20t15 08 00 758148 00 00 to datawarehouse prefect flows Flow URL <https cloud prefect io gv energy group flow 4c3ff3d7 ac87 43f4 bfe8 ded70fb20b22> ID b3c85743 d98e 4fd4 9dde f24ec1a8a0a0 Project Data Platform Develop Labels lt Flow name= dbt smartreporting gt 2022 05 20 17 08 03+0200 INFO prefect Azure | Uploading dbt datawarehouse daily 2022 05 20t15 08 00 758148 00 00 to datawarehouse prefect flows Traceback most recent call last ```

Bob De Schutter · Answer

so the second name is clearly different but prefect still tries to upload it to a blob that has the name of the previous flow for some reason thinking face

Kevin Kho · Answer

Can you try printing `flow storage` maybe

Bob De Schutter · Answer

that just gives `Storage Azure` twice also storage blob name and storage flows are empty

Bob De Schutter · Answer

ok I found it I was creating the storage object outside of the loop I changed it so I create it again on every loop iteration and now it works correclty

Kevin Kho · Answer

Ah ok glad you figured it out

Bob De Schutter · Answer

I didn t think about it but it s logical However it s strange that this worked before thinking face

Bob De Schutter · Answer

thanks for your help anyway slightly smiling face

Kevin · Answer

< Kevin Kho> < Bob De Schutter> i m running into a similar issue It started randomly i think as it had worked a few times before I am simply trying to register flows using globbing ```prefect register p $GITHUB WORKSPACE orchestrate flows project $PREFECT PROJECT```

Kevin · Answer

The first flow registration succeeds but all following flows fail due to ErrorCode BlobAlreadyExists its trying to register all of the flows using the same name from the first registered flow

Kevin · Answer

```Collecting flows Processing home runner work analytics analytics orchestrate flows salesforce salesforce py Building `Azure` storage 2022 07 07 17 52 24+0000 INFO prefect Azure | Uploading salesforce 2022 07 07t17 52 24 299152 00 00 to prefect Registering salesforce Done ID 69ac0015 cccc 43d0 9187 a1be71fc2238 Version 5 Processing home runner work analytics analytics orchestrate flows deployment deploy new source py Building `Azure` storage 2022 07 07 17 52 26+0000 INFO prefect Azure | Uploading salesforce 2022 07 07t17 52 24 299152 00 00 to prefect Error building storage Traceback most recent call last File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages prefect cli build register py line 463 in build and register storage build File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages prefect storage azure py line 155 in build client upload blob data overwrite=self overwrite File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure core tracing decorator py line 78 in wrapper use tracer return func args kwargs File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure storage blob blob client py line 728 in upload blob return upload block blob options File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure storage blob upload helpers py line 177 in upload block blob process storage error error File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure storage blob shared response handlers py line 181 in process storage error exec raise error from None pylint disable=exec used nosec File lt string gt line 1 in lt module gt File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure storage blob upload helpers py line 115 in upload block blob kwargs File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure core tracing decorator py line 78 in wrapper use tracer return func args kwargs File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure storage blob generated operations block blob operations py line 801 in upload map error status code=response status code response=response error map=error map File opt hostedtoolcache Python 3 7 13 x64 lib python3 7 site packages azure core exceptions py line 107 in map error raise error azure core exceptions ResourceExistsError The specified blob already exists RequestId 80beb2d6 701e 0067 262a 925d68000000 Time 2022 07 07T17 52 26 1641349Z ErrorCode BlobAlreadyExists Content lt xml version= 1 0 encoding= utf 8 gt lt Error gt lt Code gt BlobAlreadyExists lt Code gt lt Message gt The specified blob already exists RequestId 80beb2d6 701e 0067 262a 925d68000000 Time 2022 07 07T17 52 26 1641349Z lt Message gt lt Error gt ```

Kevin Kho · Answer

Can I see an example of how you attach sotrage

Kevin · Answer

I ended up looping through the files and registering them individually with no issues

Kevin · Answer

in it s own module i have the following ```from prefect storage import Azure from prefect run configs import KubernetesRun import os azure store = Azure container= prefect connection string secret= AZ CONNECTION STRING ```

Kevin · Answer

And then import that into each flow and use it like ```with Flow salesforce run config=kubernetes run storage=azure store as flow ```