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 8DemoETLFlow - alpha
could that be an issue? Also they're both registered to the same project and have different schedulesvgi
for the flows, forcing the flows to overwrite each other. Clearing the prefect postgres directory and restarting prefect server resolved the issue.Jim Crist-Harif
08/18/2020, 7:35 PM