Sven Teresniak
07/20/2020, 12:14 PMflow1
:
#!/usr/bin/env python
# coding: utf-8
import prefect
from prefect import task, Flow
from prefect.environments.storage.local import Local
@task
def task1():
prefect.context.get("logger").info("hello, task1")
with Flow("flow1", storage=Local(directory="/flows/.prefect/flows")) as flow:
task1()
if __name__ == "__main__":
flow.register()
And here is flow2
:
#!/usr/bin/env python
# coding: utf-8
import prefect
from prefect import task, Flow
from prefect.tasks.prefect import FlowRunTask
from prefect.environments.storage.local import Local
@task
def task2():
prefect.context.get("logger").info("hello, task2")
with Flow("flow2", storage=Local(directory="/flows/.prefect/flows")) as flow:
flow1_run = FlowRunTask(name="flow1-runner", flow_name="flow1", wait=True)
task2.set_downstream(flow1_run)
if __name__ == "__main__":
flow.register()
When I now run the flow2
I get an error: