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: