alex
08/18/2020, 6:22 PMJim Crist-Harif
08/18/2020, 6:30 PMalex
08/18/2020, 7:07 PM``` demo.py
def get_flow(
flow_name,
all_feeds,
):
with Flow(flow_name, state_handlers=state_handlers) as flow:
for feed in all_feeds:
if feed.do_x:
x = DoXTask(
feed,
upstream_tasks=[latest],
task_args=dict(name=f"XTask: {feed.feed_name}"),
)
latest = x
# build more conditional logic, use latest as upstream
return flow
name = CLI[0]
feeds = build_feeds(name) # list of classes
flow = get_flow(name, feeds)
flow.register()
```
I'm running python demo.py alpha
and python demo.py bravo
I'm expecting 2 flows alpha and bravo but instead, alpha is getting replaced by bravo (with a version increase)Jim Crist-Harif
08/18/2020, 7:12 PMget_flow
is indeed a flow with different names for different runs, you should have two separate flows registered (flows are unique per (project, name)
tuple). Are you sure your flows are being generated with different names?alex
08/18/2020, 7:24 PMalpha
with version 7. The current one is bravo
with version 8alex
08/18/2020, 7:26 PMDemoETLFlow - alpha
could that be an issue? Also they're both registered to the same project and have different schedulesalex
08/18/2020, 7:34 PMvgi
for the flows, forcing the flows to overwrite each other. Clearing the prefect postgres directory and restarting prefect server resolved the issue.alex
08/18/2020, 7:34 PMJim Crist-Harif
08/18/2020, 7:35 PM