We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
DockerEnvironment push/pull needs newline
Can print nothing to just move the cursor down a single line so output doesn't overwrite push/pull progress bar

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/environments/docker.py`
Content:
```
1 # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/beta-eula
2 
3 import filecmp
4 import json
5 import logging
6 import os
7 import shutil
8 import tempfile
9 import textwrap
10 from typing import Iterable, List
11 import uuid
12 
13 import docker
14 
15 import prefect
16 from prefect.environments import Environment
17 from prefect.environments import LocalEnvironment
18 
19 
20 class DockerEnvironment(Environment):
21     """
22     This is a base environment which takes a flow, serializes it into a LocalEnvironment,
23     and places it inside of a Docker image. This image is then used in any environment
24     which depends on using Docker containers (e.g. the Kubernetes environments).
25 
26     Args:
27         - base_image (string): the base image for this environment (e.g. `python:3.6`)
28         - registry_url (string, optional): URL of a registry to push the image to; image will not be pushed if not provided
29         - python_dependencies (List[str], optional): list of pip installable dependencies for the image
30         - image_name (string, optional): name of the image to use when building, defaults to a UUID
31         - image_tag (string, optional): tag of the image to use when building, defaults to a UUID
32         - env_vars (dict, optional): a dictionary of environment variables to use when building
33         - files (dict, optional): a dictionary of files to copy into the image when building
34     """
35 
36     def __init__(
37         self,
38         base_image: str,
39         registry_url: str = None,
40         python_dependencies: List[str] = None,
41         image_name: str = None,
42         image_tag: str = None,
43         env_vars: dict = None,
44         files: dict = None,
45     ) -> None:
46         self.base_image = base_image
47         self.registry_url = registry_url
48         self.image_name = image_name
49         self.image_tag = image_tag
50         self.python_dependencies = python_dependencies or []
51         self.env_vars = env_vars or {}
52         self.files = files or {}
53         not_absolute = [
54             file_path for file_path in self.files if not os.path.isabs(file_path)
55         ]
56         if not_absolute:
57             raise ValueError(
58                 "Provided paths {} are not absolute file paths, please provide absolute paths only.".format(
59                     ", ".join(not_absolute)
60                 )
61             )
62 
63     def _parse_generator_output(self, generator: Iterable) -> None:
64         """
65         Parses and writes a Docker command's output to stdout
66         """
67         for item in generator:
68             item = item.decode("utf-8")
69             for line in item.split("\n"):
70                 if line:
71                     output = json.loads(line).get("stream")
72                     if output and output != "\n":
73                         print(output.strip("\n"))
74 
75     def build(
76         self, flow: "prefect.Flow", push: bool = True
77     ) -> "prefect.environments.DockerEnvironment":
78         """
79         Build the Docker image. Returns a DockerEnvironment with the appropriate
80         image_name and image_tag set.
81 
82         Args:
83             - flow (prefect.Flow): Flow to be placed the image
84             - push (bool): Whether or not to push to registry after build
85 
86         Returns:
87             - DockerEnvironment: a DockerEnvironment that represents the provided flow.
88         """
89 
90         image_name, image_tag = self.build_image(flow=flow, push=push)
91 
92         return DockerEnvironment(
93             base_image=self.base_image,
94             registry_url=self.registry_url,
95             image_name=image_name,
96             image_tag=image_tag,
97         )
98 
99     def build_image(self, flow: "prefect.Flow", push: bool = True) -> tuple:
100         """
101         Build the Docker image using the docker python library. Optionally pushes the
102         image if both `push`=`True` and `self.registry_url` is set.
103 
104         Args:
105             - flow (prefect.Flow): Flow to be placed the image
106             - push (bool): Whether or not to push to registry after build
107 
108         Returns:
109             - tuple: `image_name`, `image_tag` (strings)
110         """
111         image_name = str(uuid.uuid4())
112         image_tag = str(uuid.uuid4())
113 
114         with tempfile.TemporaryDirectory() as tempdir:
115 
116             self.pull_image()
117 
118             self.create_dockerfile(flow=flow, directory=tempdir)
119 
120             client = docker.APIClient(base_url="unix://var/run/docker.sock")
121 
122             if self.registry_url:
123                 full_name = os.path.join(self.registry_url, image_name)
124             elif push is True:
125                 raise ValueError(
126                     "This environment has no `registry_url`, and cannot be pushed."
127                 )
128             else:
129                 full_name = image_name
130 
131             logging.info("Building the flow's container environment...")
132             output = client.build(
133                 path=tempdir, tag="{}:{}".format(full_name, image_tag), forcerm=True
134             )
135             self._parse_generator_output(output)
136 
137             if push:
138                 self.push_image(full_name, image_tag)
139 
140                 # Remove the image locally after being pushed
141                 client.remove_image(
142                     image="{}:{}".format(full_name, image_tag), force=True
143                 )
144 
145             return image_name, image_tag
146 
147     def pull_image(self) -> None:
148         """Pull the image specified so it can be built.
149 
150         In order for the docker python library to use a base image it must be pulled
151         from either the main docker registry or a separate registry that must be set as
152         `registry_url` on this class.
153         """
154         client = docker.APIClient(base_url="unix://var/run/docker.sock")
155 
156         output = client.pull(self.base_image, stream=True, decode=True)
157         for line in output:
158             if line.get("progress"):
159                 print(line.get("status"), line.get("progress"), end="\r")
160 
161     def push_image(self, image_name: str, image_tag: str) -> None:
162         """Push this environment to a registry
163 
164         Args:
165             - image_name (str): Name for the image
166             - image_tag (str): Tag for the image
167         """
168         client = docker.APIClient(base_url="unix://var/run/docker.sock")
169 
170         logging.info("Pushing image to the registry...")
171 
172         output = client.push(image_name, tag=image_tag, stream=True, decode=True)
173         for line in output:
174             if line.get("progress"):
175                 print(line.get("status"), line.get("progress"), end="\r")
176 
177     def create_dockerfile(self, flow: "prefect.Flow", directory: str = None) -> None:
178         """Creates a dockerfile to use as the container.
179 
180         In order for the docker python library to build a container it needs a
181         Dockerfile that it can use to define the container. This function takes the
182         image and python_dependencies then writes them to a file called Dockerfile.
183 
184         *Note*: if `files` are added to this container, they will be copied to this directory as well.
185 
186         Args:
187             - flow (Flow): the flow that the container will run
188             - directory (str, optional): A directory where the Dockerfile will be created,
189                 if no directory is specified is will be created in the current working directory
190         """
191 
192         directory = directory or "./"
193 
194         with open(os.path.join(directory, "Dockerfile"), "w+") as dockerfile:
195 
196             # Generate RUN pip install commands for python dependencies
197             pip_installs = ""
198             if self.python_dependencies:
199                 for dependency in self.python_dependencies:
200                     pip_installs += "RUN pip install {}\n".format(dependency)
201 
202             env_vars = ""
203             if self.env_vars:
204                 white_space = " " * 20
205                 env_vars = "ENV " + " \ \n{}".format(white_space).join(
206                     "{k}={v}".format(k=k, v=v) for k, v in self.env_vars.items()
207                 )
208 
209             copy_files = ""
210             if self.files:
211                 for src, dest in self.files.items():
212                     fname = os.path.basename(src)
213                     full_fname = os.path.join(directory, fname)
214                     if (
215                         os.path.exists(full_fname)
216                         and filecmp.cmp(src, full_fname) is False
217                     ):
218                         raise ValueError(
219                             "File {fname} already exists in {directory}".format(
220                                 fname=full_fname, directory=directory
221                             )
222                         )
223                     else:
224                         shutil.copy2(src, full_fname)
225                     copy_files += "COPY {fname} {dest}\n".format(fname=fname, dest=dest)
226 
227             # Create a LocalEnvironment to run the flow
228             # the local environment will be placed in the container and run when the container
229             # runs
230             local_environment = LocalEnvironment().build(flow=flow)
231             flow_path = os.path.join(directory, "flow_env.prefect")
232             local_environment.to_file(flow_path)
233 
234             # Due to prefect being a private repo it currently will require a
235             # personal access token. Once pip installable this will change and there won't
236             # be a need for the personal access token or git anymore.
237             # *Note*: this currently prevents alpine images from being used
238 
239             file_contents = textwrap.dedent(
240                 """\
241                 FROM {base_image}
242 
243                 RUN apt-get -qq -y update && apt-get -qq -y install --no-install-recommends --no-install-suggests git
244 
245                 RUN pip install pip --upgrade
246                 RUN pip install wheel
247                 {pip_installs}
248 
249                 RUN mkdir /root/.prefect/
250                 COPY flow_env.prefect /root/.prefect/flow_env.prefect
251                 {copy_files}
252 
253                 ENV PREFECT_ENVIRONMENT_FILE="/root/.prefect/flow_env.prefect"
254                 ENV PREFECT__USER_CONFIG_PATH="/root/.prefect/config.toml"
255                 {env_vars}
256 
257                 RUN git clone https://{access_token}@github.com/PrefectHQ/prefect.git
258                 RUN pip install ./prefect
259                 """.format(
260                     base_image=self.base_image,
261                     pip_installs=pip_installs,
262                     copy_files=copy_files,
263                     env_vars=env_vars,
264                     access_token=os.getenv("PERSONAL_ACCESS_TOKEN"),
265                 )
266             )
267 
268             dockerfile.write(file_contents)
269
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/environments/docker.py b/src/prefect/environments/docker.py
--- a/src/prefect/environments/docker.py
+++ b/src/prefect/environments/docker.py
@@ -157,6 +157,7 @@
         for line in output:
             if line.get("progress"):
                 print(line.get("status"), line.get("progress"), end="\r")
+        print("")
 
     def push_image(self, image_name: str, image_tag: str) -> None:
         """Push this environment to a registry
@@ -173,6 +174,7 @@
         for line in output:
             if line.get("progress"):
                 print(line.get("status"), line.get("progress"), end="\r")
+        print("")
 
     def create_dockerfile(self, flow: "prefect.Flow", directory: str = None) -> None:
         """Creates a dockerfile to use as the container.

{"golden_diff": "diff --git a/src/prefect/environments/docker.py b/src/prefect/environments/docker.py\n--- a/src/prefect/environments/docker.py\n+++ b/src/prefect/environments/docker.py\n@@ -157,6 +157,7 @@\n         for line in output:\n             if line.get(\"progress\"):\n                 print(line.get(\"status\"), line.get(\"progress\"), end=\"\\r\")\n+        print(\"\")\n \n     def push_image(self, image_name: str, image_tag: str) -> None:\n         \"\"\"Push this environment to a registry\n@@ -173,6 +174,7 @@\n         for line in output:\n             if line.get(\"progress\"):\n                 print(line.get(\"status\"), line.get(\"progress\"), end=\"\\r\")\n+        print(\"\")\n \n     def create_dockerfile(self, flow: \"prefect.Flow\", directory: str = None) -> None:\n         \"\"\"Creates a dockerfile to use as the container.\n", "issue": "DockerEnvironment push/pull needs newline\nCan print nothing to just move the cursor down a single line so output doesn't overwrite push/pull progress bar\n", "before_files": [{"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/beta-eula\n\nimport filecmp\nimport json\nimport logging\nimport os\nimport shutil\nimport tempfile\nimport textwrap\nfrom typing import Iterable, List\nimport uuid\n\nimport docker\n\nimport prefect\nfrom prefect.environments import Environment\nfrom prefect.environments import LocalEnvironment\n\n\nclass DockerEnvironment(Environment):\n    \"\"\"\n    This is a base environment which takes a flow, serializes it into a LocalEnvironment,\n    and places it inside of a Docker image. This image is then used in any environment\n    which depends on using Docker containers (e.g. the Kubernetes environments).\n\n    Args:\n        - base_image (string): the base image for this environment (e.g. `python:3.6`)\n        - registry_url (string, optional): URL of a registry to push the image to; image will not be pushed if not provided\n        - python_dependencies (List[str], optional): list of pip installable dependencies for the image\n        - image_name (string, optional): name of the image to use when building, defaults to a UUID\n        - image_tag (string, optional): tag of the image to use when building, defaults to a UUID\n        - env_vars (dict, optional): a dictionary of environment variables to use when building\n        - files (dict, optional): a dictionary of files to copy into the image when building\n    \"\"\"\n\n    def __init__(\n        self,\n        base_image: str,\n        registry_url: str = None,\n        python_dependencies: List[str] = None,\n        image_name: str = None,\n        image_tag: str = None,\n        env_vars: dict = None,\n        files: dict = None,\n    ) -> None:\n        self.base_image = base_image\n        self.registry_url = registry_url\n        self.image_name = image_name\n        self.image_tag = image_tag\n        self.python_dependencies = python_dependencies or []\n        self.env_vars = env_vars or {}\n        self.files = files or {}\n        not_absolute = [\n            file_path for file_path in self.files if not os.path.isabs(file_path)\n        ]\n        if not_absolute:\n            raise ValueError(\n                \"Provided paths {} are not absolute file paths, please provide absolute paths only.\".format(\n                    \", \".join(not_absolute)\n                )\n            )\n\n    def _parse_generator_output(self, generator: Iterable) -> None:\n        \"\"\"\n        Parses and writes a Docker command's output to stdout\n        \"\"\"\n        for item in generator:\n            item = item.decode(\"utf-8\")\n            for line in item.split(\"\\n\"):\n                if line:\n                    output = json.loads(line).get(\"stream\")\n                    if output and output != \"\\n\":\n                        print(output.strip(\"\\n\"))\n\n    def build(\n        self, flow: \"prefect.Flow\", push: bool = True\n    ) -> \"prefect.environments.DockerEnvironment\":\n        \"\"\"\n        Build the Docker image. Returns a DockerEnvironment with the appropriate\n        image_name and image_tag set.\n\n        Args:\n            - flow (prefect.Flow): Flow to be placed the image\n            - push (bool): Whether or not to push to registry after build\n\n        Returns:\n            - DockerEnvironment: a DockerEnvironment that represents the provided flow.\n        \"\"\"\n\n        image_name, image_tag = self.build_image(flow=flow, push=push)\n\n        return DockerEnvironment(\n            base_image=self.base_image,\n            registry_url=self.registry_url,\n            image_name=image_name,\n            image_tag=image_tag,\n        )\n\n    def build_image(self, flow: \"prefect.Flow\", push: bool = True) -> tuple:\n        \"\"\"\n        Build the Docker image using the docker python library. Optionally pushes the\n        image if both `push`=`True` and `self.registry_url` is set.\n\n        Args:\n            - flow (prefect.Flow): Flow to be placed the image\n            - push (bool): Whether or not to push to registry after build\n\n        Returns:\n            - tuple: `image_name`, `image_tag` (strings)\n        \"\"\"\n        image_name = str(uuid.uuid4())\n        image_tag = str(uuid.uuid4())\n\n        with tempfile.TemporaryDirectory() as tempdir:\n\n            self.pull_image()\n\n            self.create_dockerfile(flow=flow, directory=tempdir)\n\n            client = docker.APIClient(base_url=\"unix://var/run/docker.sock\")\n\n            if self.registry_url:\n                full_name = os.path.join(self.registry_url, image_name)\n            elif push is True:\n                raise ValueError(\n                    \"This environment has no `registry_url`, and cannot be pushed.\"\n                )\n            else:\n                full_name = image_name\n\n            logging.info(\"Building the flow's container environment...\")\n            output = client.build(\n                path=tempdir, tag=\"{}:{}\".format(full_name, image_tag), forcerm=True\n            )\n            self._parse_generator_output(output)\n\n            if push:\n                self.push_image(full_name, image_tag)\n\n                # Remove the image locally after being pushed\n                client.remove_image(\n                    image=\"{}:{}\".format(full_name, image_tag), force=True\n                )\n\n            return image_name, image_tag\n\n    def pull_image(self) -> None:\n        \"\"\"Pull the image specified so it can be built.\n\n        In order for the docker python library to use a base image it must be pulled\n        from either the main docker registry or a separate registry that must be set as\n        `registry_url` on this class.\n        \"\"\"\n        client = docker.APIClient(base_url=\"unix://var/run/docker.sock\")\n\n        output = client.pull(self.base_image, stream=True, decode=True)\n        for line in output:\n            if line.get(\"progress\"):\n                print(line.get(\"status\"), line.get(\"progress\"), end=\"\\r\")\n\n    def push_image(self, image_name: str, image_tag: str) -> None:\n        \"\"\"Push this environment to a registry\n\n        Args:\n            - image_name (str): Name for the image\n            - image_tag (str): Tag for the image\n        \"\"\"\n        client = docker.APIClient(base_url=\"unix://var/run/docker.sock\")\n\n        logging.info(\"Pushing image to the registry...\")\n\n        output = client.push(image_name, tag=image_tag, stream=True, decode=True)\n        for line in output:\n            if line.get(\"progress\"):\n                print(line.get(\"status\"), line.get(\"progress\"), end=\"\\r\")\n\n    def create_dockerfile(self, flow: \"prefect.Flow\", directory: str = None) -> None:\n        \"\"\"Creates a dockerfile to use as the container.\n\n        In order for the docker python library to build a container it needs a\n        Dockerfile that it can use to define the container. This function takes the\n        image and python_dependencies then writes them to a file called Dockerfile.\n\n        *Note*: if `files` are added to this container, they will be copied to this directory as well.\n\n        Args:\n            - flow (Flow): the flow that the container will run\n            - directory (str, optional): A directory where the Dockerfile will be created,\n                if no directory is specified is will be created in the current working directory\n        \"\"\"\n\n        directory = directory or \"./\"\n\n        with open(os.path.join(directory, \"Dockerfile\"), \"w+\") as dockerfile:\n\n            # Generate RUN pip install commands for python dependencies\n            pip_installs = \"\"\n            if self.python_dependencies:\n                for dependency in self.python_dependencies:\n                    pip_installs += \"RUN pip install {}\\n\".format(dependency)\n\n            env_vars = \"\"\n            if self.env_vars:\n                white_space = \" \" * 20\n                env_vars = \"ENV \" + \" \\ \\n{}\".format(white_space).join(\n                    \"{k}={v}\".format(k=k, v=v) for k, v in self.env_vars.items()\n                )\n\n            copy_files = \"\"\n            if self.files:\n                for src, dest in self.files.items():\n                    fname = os.path.basename(src)\n                    full_fname = os.path.join(directory, fname)\n                    if (\n                        os.path.exists(full_fname)\n                        and filecmp.cmp(src, full_fname) is False\n                    ):\n                        raise ValueError(\n                            \"File {fname} already exists in {directory}\".format(\n                                fname=full_fname, directory=directory\n                            )\n                        )\n                    else:\n                        shutil.copy2(src, full_fname)\n                    copy_files += \"COPY {fname} {dest}\\n\".format(fname=fname, dest=dest)\n\n            # Create a LocalEnvironment to run the flow\n            # the local environment will be placed in the container and run when the container\n            # runs\n            local_environment = LocalEnvironment().build(flow=flow)\n            flow_path = os.path.join(directory, \"flow_env.prefect\")\n            local_environment.to_file(flow_path)\n\n            # Due to prefect being a private repo it currently will require a\n            # personal access token. Once pip installable this will change and there won't\n            # be a need for the personal access token or git anymore.\n            # *Note*: this currently prevents alpine images from being used\n\n            file_contents = textwrap.dedent(\n                \"\"\"\\\n                FROM {base_image}\n\n                RUN apt-get -qq -y update && apt-get -qq -y install --no-install-recommends --no-install-suggests git\n\n                RUN pip install pip --upgrade\n                RUN pip install wheel\n                {pip_installs}\n\n                RUN mkdir /root/.prefect/\n                COPY flow_env.prefect /root/.prefect/flow_env.prefect\n                {copy_files}\n\n                ENV PREFECT_ENVIRONMENT_FILE=\"/root/.prefect/flow_env.prefect\"\n                ENV PREFECT__USER_CONFIG_PATH=\"/root/.prefect/config.toml\"\n                {env_vars}\n\n                RUN git clone https://{access_token}@github.com/PrefectHQ/prefect.git\n                RUN pip install ./prefect\n                \"\"\".format(\n                    base_image=self.base_image,\n                    pip_installs=pip_installs,\n                    copy_files=copy_files,\n                    env_vars=env_vars,\n                    access_token=os.getenv(\"PERSONAL_ACCESS_TOKEN\"),\n                )\n            )\n\n            dockerfile.write(file_contents)\n", "path": "src/prefect/environments/docker.py"}], "after_files": [{"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/beta-eula\n\nimport filecmp\nimport json\nimport logging\nimport os\nimport shutil\nimport tempfile\nimport textwrap\nfrom typing import Iterable, List\nimport uuid\n\nimport docker\n\nimport prefect\nfrom prefect.environments import Environment\nfrom prefect.environments import LocalEnvironment\n\n\nclass DockerEnvironment(Environment):\n    \"\"\"\n    This is a base environment which takes a flow, serializes it into a LocalEnvironment,\n    and places it inside of a Docker image. This image is then used in any environment\n    which depends on using Docker containers (e.g. the Kubernetes environments).\n\n    Args:\n        - base_image (string): the base image for this environment (e.g. `python:3.6`)\n        - registry_url (string, optional): URL of a registry to push the image to; image will not be pushed if not provided\n        - python_dependencies (List[str], optional): list of pip installable dependencies for the image\n        - image_name (string, optional): name of the image to use when building, defaults to a UUID\n        - image_tag (string, optional): tag of the image to use when building, defaults to a UUID\n        - env_vars (dict, optional): a dictionary of environment variables to use when building\n        - files (dict, optional): a dictionary of files to copy into the image when building\n    \"\"\"\n\n    def __init__(\n        self,\n        base_image: str,\n        registry_url: str = None,\n        python_dependencies: List[str] = None,\n        image_name: str = None,\n        image_tag: str = None,\n        env_vars: dict = None,\n        files: dict = None,\n    ) -> None:\n        self.base_image = base_image\n        self.registry_url = registry_url\n        self.image_name = image_name\n        self.image_tag = image_tag\n        self.python_dependencies = python_dependencies or []\n        self.env_vars = env_vars or {}\n        self.files = files or {}\n        not_absolute = [\n            file_path for file_path in self.files if not os.path.isabs(file_path)\n        ]\n        if not_absolute:\n            raise ValueError(\n                \"Provided paths {} are not absolute file paths, please provide absolute paths only.\".format(\n                    \", \".join(not_absolute)\n                )\n            )\n\n    def _parse_generator_output(self, generator: Iterable) -> None:\n        \"\"\"\n        Parses and writes a Docker command's output to stdout\n        \"\"\"\n        for item in generator:\n            item = item.decode(\"utf-8\")\n            for line in item.split(\"\\n\"):\n                if line:\n                    output = json.loads(line).get(\"stream\")\n                    if output and output != \"\\n\":\n                        print(output.strip(\"\\n\"))\n\n    def build(\n        self, flow: \"prefect.Flow\", push: bool = True\n    ) -> \"prefect.environments.DockerEnvironment\":\n        \"\"\"\n        Build the Docker image. Returns a DockerEnvironment with the appropriate\n        image_name and image_tag set.\n\n        Args:\n            - flow (prefect.Flow): Flow to be placed the image\n            - push (bool): Whether or not to push to registry after build\n\n        Returns:\n            - DockerEnvironment: a DockerEnvironment that represents the provided flow.\n        \"\"\"\n\n        image_name, image_tag = self.build_image(flow=flow, push=push)\n\n        return DockerEnvironment(\n            base_image=self.base_image,\n            registry_url=self.registry_url,\n            image_name=image_name,\n            image_tag=image_tag,\n        )\n\n    def build_image(self, flow: \"prefect.Flow\", push: bool = True) -> tuple:\n        \"\"\"\n        Build the Docker image using the docker python library. Optionally pushes the\n        image if both `push`=`True` and `self.registry_url` is set.\n\n        Args:\n            - flow (prefect.Flow): Flow to be placed the image\n            - push (bool): Whether or not to push to registry after build\n\n        Returns:\n            - tuple: `image_name`, `image_tag` (strings)\n        \"\"\"\n        image_name = str(uuid.uuid4())\n        image_tag = str(uuid.uuid4())\n\n        with tempfile.TemporaryDirectory() as tempdir:\n\n            self.pull_image()\n\n            self.create_dockerfile(flow=flow, directory=tempdir)\n\n            client = docker.APIClient(base_url=\"unix://var/run/docker.sock\")\n\n            if self.registry_url:\n                full_name = os.path.join(self.registry_url, image_name)\n            elif push is True:\n                raise ValueError(\n                    \"This environment has no `registry_url`, and cannot be pushed.\"\n                )\n            else:\n                full_name = image_name\n\n            logging.info(\"Building the flow's container environment...\")\n            output = client.build(\n                path=tempdir, tag=\"{}:{}\".format(full_name, image_tag), forcerm=True\n            )\n            self._parse_generator_output(output)\n\n            if push:\n                self.push_image(full_name, image_tag)\n\n                # Remove the image locally after being pushed\n                client.remove_image(\n                    image=\"{}:{}\".format(full_name, image_tag), force=True\n                )\n\n            return image_name, image_tag\n\n    def pull_image(self) -> None:\n        \"\"\"Pull the image specified so it can be built.\n\n        In order for the docker python library to use a base image it must be pulled\n        from either the main docker registry or a separate registry that must be set as\n        `registry_url` on this class.\n        \"\"\"\n        client = docker.APIClient(base_url=\"unix://var/run/docker.sock\")\n\n        output = client.pull(self.base_image, stream=True, decode=True)\n        for line in output:\n            if line.get(\"progress\"):\n                print(line.get(\"status\"), line.get(\"progress\"), end=\"\\r\")\n        print(\"\")\n\n    def push_image(self, image_name: str, image_tag: str) -> None:\n        \"\"\"Push this environment to a registry\n\n        Args:\n            - image_name (str): Name for the image\n            - image_tag (str): Tag for the image\n        \"\"\"\n        client = docker.APIClient(base_url=\"unix://var/run/docker.sock\")\n\n        logging.info(\"Pushing image to the registry...\")\n\n        output = client.push(image_name, tag=image_tag, stream=True, decode=True)\n        for line in output:\n            if line.get(\"progress\"):\n                print(line.get(\"status\"), line.get(\"progress\"), end=\"\\r\")\n        print(\"\")\n\n    def create_dockerfile(self, flow: \"prefect.Flow\", directory: str = None) -> None:\n        \"\"\"Creates a dockerfile to use as the container.\n\n        In order for the docker python library to build a container it needs a\n        Dockerfile that it can use to define the container. This function takes the\n        image and python_dependencies then writes them to a file called Dockerfile.\n\n        *Note*: if `files` are added to this container, they will be copied to this directory as well.\n\n        Args:\n            - flow (Flow): the flow that the container will run\n            - directory (str, optional): A directory where the Dockerfile will be created,\n                if no directory is specified is will be created in the current working directory\n        \"\"\"\n\n        directory = directory or \"./\"\n\n        with open(os.path.join(directory, \"Dockerfile\"), \"w+\") as dockerfile:\n\n            # Generate RUN pip install commands for python dependencies\n            pip_installs = \"\"\n            if self.python_dependencies:\n                for dependency in self.python_dependencies:\n                    pip_installs += \"RUN pip install {}\\n\".format(dependency)\n\n            env_vars = \"\"\n            if self.env_vars:\n                white_space = \" \" * 20\n                env_vars = \"ENV \" + \" \\ \\n{}\".format(white_space).join(\n                    \"{k}={v}\".format(k=k, v=v) for k, v in self.env_vars.items()\n                )\n\n            copy_files = \"\"\n            if self.files:\n                for src, dest in self.files.items():\n                    fname = os.path.basename(src)\n                    full_fname = os.path.join(directory, fname)\n                    if (\n                        os.path.exists(full_fname)\n                        and filecmp.cmp(src, full_fname) is False\n                    ):\n                        raise ValueError(\n                            \"File {fname} already exists in {directory}\".format(\n                                fname=full_fname, directory=directory\n                            )\n                        )\n                    else:\n                        shutil.copy2(src, full_fname)\n                    copy_files += \"COPY {fname} {dest}\\n\".format(fname=fname, dest=dest)\n\n            # Create a LocalEnvironment to run the flow\n            # the local environment will be placed in the container and run when the container\n            # runs\n            local_environment = LocalEnvironment().build(flow=flow)\n            flow_path = os.path.join(directory, \"flow_env.prefect\")\n            local_environment.to_file(flow_path)\n\n            # Due to prefect being a private repo it currently will require a\n            # personal access token. Once pip installable this will change and there won't\n            # be a need for the personal access token or git anymore.\n            # *Note*: this currently prevents alpine images from being used\n\n            file_contents = textwrap.dedent(\n                \"\"\"\\\n                FROM {base_image}\n\n                RUN apt-get -qq -y update && apt-get -qq -y install --no-install-recommends --no-install-suggests git\n\n                RUN pip install pip --upgrade\n                RUN pip install wheel\n                {pip_installs}\n\n                RUN mkdir /root/.prefect/\n                COPY flow_env.prefect /root/.prefect/flow_env.prefect\n                {copy_files}\n\n                ENV PREFECT_ENVIRONMENT_FILE=\"/root/.prefect/flow_env.prefect\"\n                ENV PREFECT__USER_CONFIG_PATH=\"/root/.prefect/config.toml\"\n                {env_vars}\n\n                RUN git clone https://{access_token}@github.com/PrefectHQ/prefect.git\n                RUN pip install ./prefect\n                \"\"\".format(\n                    base_image=self.base_image,\n                    pip_installs=pip_installs,\n                    copy_files=copy_files,\n                    env_vars=env_vars,\n                    access_token=os.getenv(\"PERSONAL_ACCESS_TOKEN\"),\n                )\n            )\n\n            dockerfile.write(file_contents)\n", "path": "src/prefect/environments/docker.py"}]}