Allow Local Images for Docker Storage

CHANGELOG.md

@@ -17,6 +17,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 - Allow for `SlackTask` to pull the Slack webhook URL from a custom named Secret - [#1023](https://github.com/PrefectHQ/prefect/pull/1023)
 - Raise informative errors when Docker storage push / pull fails - [#1029](https://github.com/PrefectHQ/prefect/issues/1029)
 - Standardized `__repr__`s for various classes, to remove inconsistencies  - [#617](https://github.com/PrefectHQ/prefect/issues/617)
+- Allow for use of local images in Docekr storage - [#1052](https://github.com/PrefectHQ/prefect/pull/1052)
 
 ### Task Library
 

src/prefect/environments/storage/docker.py

@@ -40,6 +40,7 @@ class Docker(Storage):
         - prefect_version (str, optional): an optional branch, tag, or commit specifying the version of prefect
             you want installed into the container; defaults to the version you are currently using or `"master"` if your version is ahead of
             the latest tag
+        - local_image(bool, optional): an optional flag whether or not to use a local docker image, if True then a pull will not be attempted
     """
 
     def __init__(
@@ -53,6 +54,7 @@ def __init__(
         files: dict = None,
         base_url: str = "unix://var/run/docker.sock",
         prefect_version: str = None,
+        local_image: bool = False,
     ) -> None:
         self.registry_url = registry_url
 
@@ -72,6 +74,7 @@ def __init__(
         self.flows = dict()  # type: Dict[str, str]
         self._flows = dict()  # type: Dict[str, "prefect.core.flow.Flow"]
         self.base_url = base_url
+        self.local_image = local_image
 
         version = prefect.__version__.split("+")
         if prefect_version is None:
@@ -214,7 +217,7 @@ def build_image(self, push: bool = True) -> tuple:
         with tempfile.TemporaryDirectory() as tempdir:
 
             # Build the dockerfile
-            if self.base_image:
+            if self.base_image and not self.local_image:
                 self.pull_image()
 
             self.create_dockerfile_object(directory=tempdir)

tests/environments/storage/test_docker_storage.py

@@ -44,6 +44,7 @@ def test_empty_docker_storage():
     assert not storage.files
     assert storage.prefect_version
     assert storage.base_url == "unix://var/run/docker.sock"
+    assert not storage.local_image
 
 
 @pytest.mark.parametrize("version_info", [(3, 5), (3, 6), (3, 7)])
@@ -78,6 +79,7 @@ def test_initialized_docker_storage():
         env_vars={"test": "1"},
         base_url="test_url",
         prefect_version="my-branch",
+        local_image=True,
     )
 
     assert storage.registry_url == "test1"
@@ -88,6 +90,7 @@ def test_initialized_docker_storage():
     assert storage.env_vars == {"test": "1"}
     assert storage.base_url == "test_url"
     assert storage.prefect_version == "my-branch"
+    assert storage.local_image
 
 
 def test_files_not_absolute_path():