Add functional API version of the Docker example

docs/guide/examples/functional_docker.md

@@ -0,0 +1,42 @@
+# Functional API: Docker Pipeline 
+
+This Prefect Flow creates a Container based on the latest prefect image, and
+executes an empty Flow inside that container.  Make sure to pull `prefecthq/prefect` prior
+to running this Flow.
+
+```python
+from prefect import Flow
+from prefect.tasks.docker import (
+    CreateContainer,
+    StartContainer,
+    GetContainerLogs,
+    WaitOnContainer,
+)
+from prefect.triggers import always_run
+
+
+## initialize tasks
+container = CreateContainer(
+    image_name="prefecthq/prefect",
+    command='''python -c "from prefect import Flow; f = Flow('empty'); f.run()"''',
+)
+start = StartContainer()
+logs = GetContainerLogs(trigger=always_run)
+status_code = WaitOnContainer()
+
+
+## set task dependencies via functional API
+
+with Flow("Run a Prefect Flow in Docker") as flow:
+    start_container = start(container_id=container)
+    code = status_code(container_id=container, upstream_tasks=[start_container])
+    collect_logs = logs(container_id=container, upstream_tasks=[code])
+
+## run flow and print logs
+flow_state = flow.run()
+
+print("=" * 30)
+print("Container Logs")
+print("=" * 30)
+print(flow_state.result[collect_logs].result)
+```

docs/guide/examples/imperative_docker.md

@@ -1,7 +1,8 @@
 # Imperative API: Docker Pipeline 
 
 This Prefect Flow creates a Container based on the latest prefect image, and
-executes an empty Flow inside that container.
+executes an empty Flow inside that container.  Make sure to pull `prefecthq/prefect` prior
+to running this Flow.
 
 ```python
 from prefect import Flow
@@ -16,7 +17,7 @@ from prefect.triggers import always_run
 
 container = CreateContainer(
     image_name="prefecthq/prefect",
-    command='''python -c "from prefect import Flow; f = Flow("empty"); f.run()"''',
+    command='''python -c "from prefect import Flow; f = Flow('empty'); f.run()"''',
 )
 start = StartContainer()
 logs = GetContainerLogs(trigger=always_run)

examples/functional_docker.py

@@ -0,0 +1,40 @@
+"""
+This Prefect Flow creates a Container based on the latest prefect image, and
+executes an empty Flow inside that container.  Make sure to pull `prefecthq/prefect` prior
+to running this Flow.
+"""
+
+from prefect import Flow
+from prefect.tasks.docker import (
+    CreateContainer,
+    StartContainer,
+    GetContainerLogs,
+    WaitOnContainer,
+)
+from prefect.triggers import always_run
+
+
+## initialize tasks
+container = CreateContainer(
+    image_name="prefecthq/prefect",
+    command='''python -c "from prefect import Flow; f = Flow('empty'); f.run()"''',
+)
+start = StartContainer()
+logs = GetContainerLogs(trigger=always_run)
+status_code = WaitOnContainer()
+
+
+## set task dependencies via functional API
+
+with Flow("Run a Prefect Flow in Docker") as flow:
+    start_container = start(container_id=container)
+    code = status_code(container_id=container, upstream_tasks=[start_container])
+    collect_logs = logs(container_id=container, upstream_tasks=[code])
+
+## run flow and print logs
+flow_state = flow.run()
+
+print("=" * 30)
+print("Container Logs")
+print("=" * 30)
+print(flow_state.result[collect_logs].result)

examples/imperative_docker.py

@@ -1,6 +1,7 @@
 """
 This Prefect Flow creates a Container based on the latest prefect image, and
-executes an empty Flow inside that container.
+executes an empty Flow inside that container.  Make sure to pull `prefecthq/prefect` prior
+to running this Flow.
 """
 
 from prefect import Flow
@@ -15,7 +16,7 @@
 
 container = CreateContainer(
     image_name="prefecthq/prefect",
-    command='''python -c "from prefect import Flow; f = Flow("empty"); f.run()"''',
+    command='''python -c "from prefect import Flow; f = Flow('empty'); f.run()"''',
 )
 start = StartContainer()
 logs = GetContainerLogs(trigger=always_run)