Repository: PrefectHQ/prefect

Issue:
File with multiple flows stored as script fails to run correct flow
When declaring multiple flows (with different names) in a file with `stored_as_script=True`, running any of the flows via an agent seems to only pick up the first flow. I believe this is caused by the following line, as it doesn't pass the `flow_name` when running `get_flow()`.

https://github.com/PrefectHQ/prefect/blob/d4ad74a819eff585b0553b9479dd853e75089489/src/prefect/cli/execute.py#L67

The `get_flow()` function calls `extract_flow_from_file()` which supports an optional parameter for `flow_name`, so maybe this should be passed through.

My main use-case for this is running flows with a big time gap, e.g. running a long running task for 8 hours and then stopping it, where it doesn't make sense to keep the flow running with a `sleep()` for that amount of time. In this case, it is nicer to keep the flow code together instead of in separate files.

Discussion:

Patch:
diff --git a/changes/pr3969.yaml b/changes/pr3969.yaml
new file mode 100644
index 0000000000..35ca8f36a0
--- /dev/null
+++ b/changes/pr3969.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Fix support for storing multiple flows in a single script in storage - [#3969](https://github.com/PrefectHQ/prefect/pull/3969)"
diff --git a/src/prefect/cli/execute.py b/src/prefect/cli/execute.py
index d2ad6a37e4..a0af786d59 100644
--- a/src/prefect/cli/execute.py
+++ b/src/prefect/cli/execute.py
@@ -64,7 +64,7 @@ def flow_run():
             secrets[secret] = PrefectSecret(name=secret).run()
 
         with prefect.context(secrets=secrets, loading_flow=True):
-            flow = storage.get_flow(storage.flows[flow_data.name])
+            flow = storage.get_flow(flow_data.name)
 
         with prefect.context(secrets=secrets):
             if flow_data.run_config is not None:
diff --git a/src/prefect/environments/execution/base.py b/src/prefect/environments/execution/base.py
index 6203b75ac4..432e49b321 100644
--- a/src/prefect/environments/execution/base.py
+++ b/src/prefect/environments/execution/base.py
@@ -169,7 +169,7 @@ def load_and_run_flow() -> None:
             secrets[secret] = prefect.tasks.secrets.PrefectSecret(name=secret).run()
 
         with prefect.context(secrets=secrets):
-            flow = storage.get_flow(storage.flows[flow_data.name])
+            flow = storage.get_flow(flow_data.name)
             flow.environment.run(flow)
     except Exception as exc:
         logger.exception("Unexpected error raised during flow run: {}".format(exc))
diff --git a/src/prefect/storage/__init__.py b/src/prefect/storage/__init__.py
index be13b55f5c..d589232b7f 100644
--- a/src/prefect/storage/__init__.py
+++ b/src/prefect/storage/__init__.py
@@ -8,7 +8,7 @@ within a given unit), and exposes the following methods and attributes:
 location of the given flow in the Storage unit (note flow uploading/saving does not happen until `build`)
 - the `__contains__(self, obj) -> bool` special method for determining whether the Storage contains a
 given Flow
-- one of `get_flow(flow_location: str)` for retrieving a way of interfacing with either `flow.run` or a
+- one of `get_flow(flow_name: str)` for retrieving a way of interfacing with either `flow.run` or a
 `FlowRunner` for the flow
 - a `build() -> Storage` method for "building" the storage. In storage options where flows are stored in
 an external service (such as S3 and the filesystem) the flows are uploaded/saved during this step
diff --git a/src/prefect/storage/azure.py b/src/prefect/storage/azure.py
index 18e50793fa..5e2bab5d46 100644
--- a/src/prefect/storage/azure.py
+++ b/src/prefect/storage/azure.py
@@ -60,27 +60,19 @@ class Azure(Storage):
         )
         super().__init__(result=result, stored_as_script=stored_as_script, **kwargs)
 
-    def get_flow(self, flow_location: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str, optional): the location of a flow within this Storage; in this case,
-                a file path where a Flow has been serialized to. Will use `blob_name` if not provided.
+            - flow_name (str): the name of the flow to return.
 
         Returns:
             - Flow: the requested flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.blob_name:
-            flow_location = self.blob_name
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         client = self._azure_block_blob_service.get_blob_client(
             container=self.container, blob=flow_location
@@ -91,7 +83,7 @@ class Azure(Storage):
         content = client.download_blob().content_as_bytes()
 
         if self.stored_as_script:
-            return extract_flow_from_file(file_contents=content)  # type: ignore
+            return extract_flow_from_file(file_contents=content, flow_name=flow_name)  # type: ignore
 
         return flow_from_bytes_pickle(content)
 
diff --git a/src/prefect/storage/base.py b/src/prefect/storage/base.py
index bcde480bac..ff00e9eb84 100644
--- a/src/prefect/storage/base.py
+++ b/src/prefect/storage/base.py
@@ -88,12 +88,12 @@ class Storage(metaclass=ABCMeta):
         """
         raise NotImplementedError()
 
-    def get_flow(self, flow_location: str) -> "prefect.core.flow.Flow":
+    def get_flow(self, flow_name: str) -> "prefect.core.flow.Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str): the location of a flow within this Storage
+            - flow_name (str): the name of the flow to return.
 
         Returns:
             - Flow: the requested flow
diff --git a/src/prefect/storage/bitbucket.py b/src/prefect/storage/bitbucket.py
index b5fbc8d9af..1bd22315aa 100644
--- a/src/prefect/storage/bitbucket.py
+++ b/src/prefect/storage/bitbucket.py
@@ -61,35 +61,22 @@ class Bitbucket(Storage):
 
         super().__init__(**kwargs)
 
-    def get_flow(self, flow_location: str = None, ref: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
-        If the Flow is not found an error will be logged and `None` will be returned.
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str): the location of a flow within this Storage; in this case,
-                a file path on a repository where a Flow file has been committed. Will use `path` if not
-                provided.
-            - ref (str, optional): a commit SHA-1 value or branch name. Defaults to 'master' if
-                not specified
+            - flow_name (str): the name of the flow to return.
 
         Returns:
-            - Flow: the requested Flow; Atlassian API retrieves raw, decoded files.
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
-            - HTTPError: if flow is unable to access the Bitbucket repository
+            - Flow: the requested flow
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.path:
-            flow_location = self.path
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
-        # Use ref argument if exists, else use attribute, else default to 'master'
-        ref = ref if ref else (self.ref if self.ref else "master")
+        # Use ref attribute if present, defaulting to "master"
+        ref = self.ref or "master"
 
         try:
             contents = self._bitbucket_client.get_content_of_file(
@@ -116,7 +103,7 @@ class Bitbucket(Storage):
                 )
                 raise
 
-        return extract_flow_from_file(file_contents=contents)
+        return extract_flow_from_file(file_contents=contents, flow_name=flow_name)
 
     def add_flow(self, flow: "Flow") -> str:
         """
diff --git a/src/prefect/storage/codecommit.py b/src/prefect/storage/codecommit.py
index f8cfd15339..5fff125f62 100644
--- a/src/prefect/storage/codecommit.py
+++ b/src/prefect/storage/codecommit.py
@@ -55,30 +55,19 @@ class CodeCommit(Storage):
 
         super().__init__(**kwargs)
 
-    def get_flow(self, flow_location: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
-        If the Flow is not found an error will be logged and `None` will be returned.
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str): the location of a flow within this Storage; in this case,
-                a file path on a repository where a Flow file has been committed. Will use `path` if not
-                provided.
+            - flow_name (str): the name of the flow to return.
 
         Returns:
-            - Flow: the requested Flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
-            - UnknownObjectException: if the flow file is unable to be retrieved
+            - Flow: the requested flow
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.path:
-            flow_location = self.path
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         client = self._boto3_client
 
@@ -97,7 +86,9 @@ class CodeCommit(Storage):
             )
             raise exc
 
-        return extract_flow_from_file(file_contents=decoded_contents)
+        return extract_flow_from_file(
+            file_contents=decoded_contents, flow_name=flow_name
+        )
 
     def add_flow(self, flow: "Flow") -> str:
         """
diff --git a/src/prefect/storage/docker.py b/src/prefect/storage/docker.py
index a0c4d7b2a4..47eedfc459 100644
--- a/src/prefect/storage/docker.py
+++ b/src/prefect/storage/docker.py
@@ -285,31 +285,22 @@ class Docker(Storage):
         self._flows[flow.name] = flow  # needed prior to build
         return flow_path
 
-    def get_flow(self, flow_location: str = None) -> "prefect.core.flow.Flow":
+    def get_flow(self, flow_name: str) -> "prefect.core.flow.Flow":
         """
-        Given a file path within this Docker container, returns the underlying Flow.
-        Note that this method should only be run _within_ the container itself.
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str, optional): the file path of a flow within this container. Will use
-                `path` if not provided.
+            - flow_name (str): the name of the flow to return.
 
         Returns:
             - Flow: the requested flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.path:
-            flow_location = self.path
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         if self.stored_as_script:
-            return extract_flow_from_file(file_path=flow_location)
+            return extract_flow_from_file(file_path=flow_location, flow_name=flow_name)
 
         with open(flow_location, "rb") as f:
             return flow_from_bytes_pickle(f.read())
diff --git a/src/prefect/storage/gcs.py b/src/prefect/storage/gcs.py
index d58244e9c1..4ad551944c 100644
--- a/src/prefect/storage/gcs.py
+++ b/src/prefect/storage/gcs.py
@@ -66,27 +66,19 @@ class GCS(Storage):
         result = GCSResult(bucket=bucket)
         super().__init__(result=result, stored_as_script=stored_as_script, **kwargs)
 
-    def get_flow(self, flow_location: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str, optional): the location of a flow within this Storage; in this case,
-                a file path where a Flow has been serialized to. Will use `key` if not provided.
+            - flow_name (str): the name of the flow to return.
 
         Returns:
             - Flow: the requested flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.key:
-            flow_location = self.key
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         bucket = self._gcs_client.get_bucket(self.bucket)
 
@@ -107,7 +99,7 @@ class GCS(Storage):
         )
 
         if self.stored_as_script:
-            return extract_flow_from_file(file_contents=content)
+            return extract_flow_from_file(file_contents=content, flow_name=flow_name)
 
         return flow_from_bytes_pickle(content)
 
diff --git a/src/prefect/storage/github.py b/src/prefect/storage/github.py
index b679489080..13aa7967bb 100644
--- a/src/prefect/storage/github.py
+++ b/src/prefect/storage/github.py
@@ -46,39 +46,26 @@ class GitHub(Storage):
 
         super().__init__(**kwargs)
 
-    def get_flow(self, flow_location: str = None, ref: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
-        If the Flow is not found an error will be logged and `None` will be returned.
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str): the location of a flow within this Storage; in this case,
-                a file path on a repository where a Flow file has been committed. Will use `path` if not
-                provided.
-            - ref (str, optional): a commit SHA-1 value or branch name. Defaults to 'master' if not
-                specified
+            - flow_name (str): the name of the flow to return.
 
         Returns:
-            - Flow: the requested Flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
-            - UnknownObjectException: if the flow file is unable to be retrieved
+            - Flow: the requested flow
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.path:
-            flow_location = self.path
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         from github import UnknownObjectException
 
         repo = self._github_client.get_repo(self.repo)
 
         try:
-            contents = repo.get_contents(flow_location, ref=ref or self.ref)
+            contents = repo.get_contents(flow_location, ref=self.ref)
             decoded_contents = contents.decoded_content
         except UnknownObjectException as exc:
             self.logger.error(
@@ -88,7 +75,9 @@ class GitHub(Storage):
             )
             raise exc
 
-        return extract_flow_from_file(file_contents=decoded_contents)
+        return extract_flow_from_file(
+            file_contents=decoded_contents, flow_name=flow_name
+        )
 
     def add_flow(self, flow: "Flow") -> str:
         """
diff --git a/src/prefect/storage/gitlab.py b/src/prefect/storage/gitlab.py
index c65816b3ab..32e6c1da35 100644
--- a/src/prefect/storage/gitlab.py
+++ b/src/prefect/storage/gitlab.py
@@ -55,35 +55,21 @@ class GitLab(Storage):
 
         super().__init__(**kwargs)
 
-    def get_flow(self, flow_location: str = None, ref: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
-        If the Flow is not found an error will be logged and `None` will be returned.
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str): the location of a flow within this Storage; in this case,
-                a file path on a repository where a Flow file has been committed. Will use `path` if not
-                provided.
-            - ref (str, optional): a commit SHA-1 value or branch name. Defaults to 'master' if
-                not specified
+            - flow_name (str): the name of the flow to return.
 
         Returns:
-            - Flow: the requested Flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
-            - UnknownObjectException: if the flow file is unable to be retrieved
+            - Flow: the requested flow
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.path:
-            flow_location = self.path
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
-        # Use ref argument if exists, else use attribute, else default to 'master'
-        ref = ref if ref else (self.ref if self.ref else "master")
+        ref = self.ref or "master"
 
         from gitlab.exceptions import GitlabAuthenticationError, GitlabGetError
 
@@ -102,7 +88,9 @@ class GitLab(Storage):
             )
             raise
 
-        return extract_flow_from_file(file_contents=contents.decode())
+        return extract_flow_from_file(
+            file_contents=contents.decode(), flow_name=flow_name
+        )
 
     def add_flow(self, flow: "Flow") -> str:
         """
diff --git a/src/prefect/storage/local.py b/src/prefect/storage/local.py
index 616a2b5396..4b9de2eefa 100644
--- a/src/prefect/storage/local.py
+++ b/src/prefect/storage/local.py
@@ -76,43 +76,32 @@ class Local(Storage):
         else:
             return []
 
-    def get_flow(self, flow_location: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object, returns the underlying Flow (if possible).
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str, optional): the location of a flow within this Storage; in this case,
-                a file path or python path where a Flow has been serialized to. Will use `path`
-                if not provided.
+            - flow_name (str): the name of the flow to return.
 
         Returns:
             - Flow: the requested flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.path:
-            flow_location = self.path
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         # check if the path given is a file path
-        try:
-            if os.path.isfile(flow_location):
-                if self.stored_as_script:
-                    return extract_flow_from_file(file_path=flow_location)
-                else:
-                    with open(flow_location, "rb") as f:
-                        return flow_from_bytes_pickle(f.read())
-            # otherwise the path is given in the module format
+        if os.path.isfile(flow_location):
+            if self.stored_as_script:
+                return extract_flow_from_file(
+                    file_path=flow_location, flow_name=flow_name
+                )
             else:
-                return extract_flow_from_module(module_str=flow_location)
-        except Exception:
-            self.logger.exception(f"Failed to load Flow from {flow_location}")
-            raise
+                with open(flow_location, "rb") as f:
+                    return flow_from_bytes_pickle(f.read())
+        # otherwise the path is given in the module format
+        else:
+            return extract_flow_from_module(module_str=flow_location)
 
     def add_flow(self, flow: "Flow") -> str:
         """
diff --git a/src/prefect/storage/s3.py b/src/prefect/storage/s3.py
index e80893db7f..a4adf8d096 100644
--- a/src/prefect/storage/s3.py
+++ b/src/prefect/storage/s3.py
@@ -66,29 +66,19 @@ class S3(Storage):
             **kwargs,
         )
 
-    def get_flow(self, flow_location: str = None) -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Given a flow_location within this Storage object or S3, returns the underlying Flow
-        (if possible).
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str, optional): the location of a flow within this Storage; in this case
-                an S3 object key where a Flow has been serialized to. Will use `key` if not provided.
+            - flow_name (str): the name of the flow to return.
 
         Returns:
-            - Flow: the requested Flow
-
-        Raises:
-            - ValueError: if the flow is not contained in this storage
-            - botocore.ClientError: if there is an issue downloading the Flow from S3
+            - Flow: the requested flow
         """
-        if flow_location:
-            if flow_location not in self.flows.values():
-                raise ValueError("Flow is not contained in this Storage")
-        elif self.key:
-            flow_location = self.key
-        else:
-            raise ValueError("No flow location provided")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+        flow_location = self.flows[flow_name]
 
         stream = io.BytesIO()
 
@@ -110,7 +100,7 @@ class S3(Storage):
         output = stream.read()
 
         if self.stored_as_script:
-            return extract_flow_from_file(file_contents=output)  # type: ignore
+            return extract_flow_from_file(file_contents=output, flow_name=flow_name)  # type: ignore
 
         return flow_from_bytes_pickle(output)
 
diff --git a/src/prefect/storage/webhook.py b/src/prefect/storage/webhook.py
index 226b2e81b5..3fe54a46f1 100644
--- a/src/prefect/storage/webhook.py
+++ b/src/prefect/storage/webhook.py
@@ -232,23 +232,19 @@ class Webhook(Storage):
 
         super().__init__(stored_as_script=stored_as_script, **kwargs)
 
-    def get_flow(self, flow_location: str = "placeholder") -> "Flow":
+    def get_flow(self, flow_name: str) -> "Flow":
         """
-        Get the flow from storage. This method will call
-        `cloudpickle.loads()` on the binary content of the flow, so it
-        should only be called in an environment with all of the flow's
-        dependencies.
+        Given a flow name within this Storage object, load and return the Flow.
 
         Args:
-            - flow_location (str): This argument is included to comply with the
-                interface used by other storage objects, but it has no meaning
-                for `Webhook` storage, since `Webhook` only corresponds to a
-                single flow. Ignore it.
+            - flow_name (str): the name of the flow to return.
 
-        Raises:
-            - requests.exceptions.HTTPError if getting the flow fails
+        Returns:
+            - Flow: the requested flow
         """
-        self.logger.info("Retrieving flow")
+        if flow_name not in self.flows:
+            raise ValueError("Flow is not contained in this Storage")
+
         req_function = self._method_to_function[self.get_flow_request_http_method]
 
         get_flow_request_kwargs = _render_dict(self.get_flow_request_kwargs)
@@ -258,7 +254,9 @@ class Webhook(Storage):
 
         if self.stored_as_script:
             flow_script_content = response.content.decode("utf-8")
-            return extract_flow_from_file(file_contents=flow_script_content)  # type: ignore
+            return extract_flow_from_file(
+                file_contents=flow_script_content, flow_name=flow_name
+            )
 
         return flow_from_bytes_pickle(response.content)
 


Test patch:
diff --git a/tests/environments/execution/test_local_environment.py b/tests/environments/execution/test_local_environment.py
index 3a2a7f86b8..7090aa106d 100644
--- a/tests/environments/execution/test_local_environment.py
+++ b/tests/environments/execution/test_local_environment.py
@@ -13,8 +13,8 @@ class DummyStorage(Local):
         self.flows[flow.name] = flow
         return flow.name
 
-    def get_flow(self, location):
-        return self.flows[location]
+    def get_flow(self, flow_name):
+        return self.flows[flow_name]
 
 
 def test_create_environment():
diff --git a/tests/serialization/test_storage.py b/tests/serialization/test_storage.py
index 7f82a36fd1..0eead6281e 100644
--- a/tests/serialization/test_storage.py
+++ b/tests/serialization/test_storage.py
@@ -234,12 +234,12 @@ def test_local_empty_serialize():
 def test_local_roundtrip():
     with tempfile.TemporaryDirectory() as tmpdir:
         s = storage.Local(directory=tmpdir, secrets=["AUTH"])
-        flow_loc = s.add_flow(prefect.Flow("test"))
+        s.add_flow(prefect.Flow("test"))
         serialized = LocalSchema().dump(s)
         deserialized = LocalSchema().load(serialized)
 
         assert "test" in deserialized
-        runner = deserialized.get_flow(flow_loc)
+        runner = deserialized.get_flow("test")
 
     assert runner.run().is_successful()
     assert deserialized.secrets == ["AUTH"]
diff --git a/tests/storage/test_azure_storage.py b/tests/storage/test_azure_storage.py
index d218d0291b..af9c37aa6c 100644
--- a/tests/storage/test_azure_storage.py
+++ b/tests/storage/test_azure_storage.py
@@ -166,13 +166,10 @@ def test_get_flow_azure(monkeypatch):
 
     storage = Azure(container="container")
 
-    with pytest.raises(ValueError):
-        storage.get_flow()
-
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    assert storage.get_flow(flow_location)
+    assert storage.get_flow(f.name)
     assert client.download_blob.called
     assert f.name in storage
 
@@ -191,7 +188,7 @@ def test_get_flow_azure_bucket_key(monkeypatch):
     assert f.name not in storage
     flow_location = storage.add_flow(f)
 
-    assert storage.get_flow(flow_location)
+    assert storage.get_flow(f.name)
     assert service.get_blob_client.call_args[1]["container"] == "container"
     assert service.get_blob_client.call_args[1]["blob"] == flow_location
 
@@ -208,9 +205,9 @@ def test_get_flow_azure_runs(monkeypatch):
     storage = Azure(container="container")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
     assert client.download_blob.called
     assert f.name in storage
 
@@ -229,19 +226,19 @@ def test_get_flow_from_file_azure_runs(monkeypatch):
 
     f = Flow("test")
 
+    extract_flow_from_file = MagicMock(return_value=f)
     monkeypatch.setattr(
-        "prefect.storage.azure.extract_flow_from_file",
-        MagicMock(return_value=f),
+        "prefect.storage.azure.extract_flow_from_file", extract_flow_from_file
     )
 
     storage = Azure(container="container", stored_as_script=True)
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
     assert client.download_blob.called
-    assert f.name in storage
+    assert extract_flow_from_file.call_args[1]["flow_name"] == f.name
 
     assert isinstance(new_flow, Flow)
     assert new_flow.name == "test"
diff --git a/tests/storage/test_bitbucket_storage.py b/tests/storage/test_bitbucket_storage.py
index 01ed573287..804cf9af90 100644
--- a/tests/storage/test_bitbucket_storage.py
+++ b/tests/storage/test_bitbucket_storage.py
@@ -86,21 +86,16 @@ def test_get_flow_bitbucket(monkeypatch):
     bitbucket = MagicMock()
     monkeypatch.setattr("prefect.utilities.git.Bitbucket", bitbucket)
 
+    extract_flow_from_file = MagicMock(return_value=f)
     monkeypatch.setattr(
-        "prefect.storage.bitbucket.extract_flow_from_file",
-        MagicMock(return_value=f),
+        "prefect.storage.bitbucket.extract_flow_from_file", extract_flow_from_file
     )
 
-    with pytest.raises(ValueError) as ex:
-        storage = Bitbucket(project="PROJECT", repo="test-repo")
-        storage.get_flow()
-
-    assert "No flow location provided" in str(ex.value)
-
     storage = Bitbucket(project="PROJECT", repo="test-repo", path="test-flow.py")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
+    assert extract_flow_from_file.call_args[1]["flow_name"] == f.name
     assert new_flow.run()
diff --git a/tests/storage/test_codecommit_storage.py b/tests/storage/test_codecommit_storage.py
index fe5349a2fa..e8acf54157 100644
--- a/tests/storage/test_codecommit_storage.py
+++ b/tests/storage/test_codecommit_storage.py
@@ -112,19 +112,17 @@ def test_get_flow_codecommit(monkeypatch):
 
     f = Flow("test")
 
+    extract_flow_from_file = MagicMock(return_value=f)
     monkeypatch.setattr(
-        "prefect.storage.github.extract_flow_from_file",
-        MagicMock(return_value=f),
+        "prefect.storage.codecommit.extract_flow_from_file",
+        extract_flow_from_file,
     )
 
-    with pytest.raises(ValueError):
-        storage = CodeCommit(repo="test/repo")
-        storage.get_flow()
-
     storage = CodeCommit(repo="test/repo", path="flow", commit="master")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
+    assert extract_flow_from_file.call_args[1]["flow_name"] == f.name
     assert new_flow.run()
diff --git a/tests/storage/test_docker_storage.py b/tests/storage/test_docker_storage.py
index bcad920016..3f3317b2b0 100644
--- a/tests/storage/test_docker_storage.py
+++ b/tests/storage/test_docker_storage.py
@@ -1,5 +1,4 @@
 import os
-import re
 import sys
 import json
 import tempfile
@@ -210,7 +209,7 @@ def test_docker_storage_allows_for_user_provided_config_locations():
 
 def test_files_not_absolute_path():
     with pytest.raises(ValueError):
-        storage = Docker(files={"test": "test"})
+        Docker(files={"test": "test"})
 
 
 def test_build_base_image(monkeypatch):
@@ -870,32 +869,20 @@ def test_docker_storage_name_registry_url_none():
     assert storage.name == "test2:test3"
 
 
-def test_docker_storage_get_flow_method():
-    with tempfile.TemporaryDirectory() as directory:
-        storage = Docker(base_image="python:3.6", prefect_directory=directory)
+def test_docker_storage_get_flow_method(tmpdir):
+    flow_dir = str(tmpdir.mkdir("flows"))
 
-        with pytest.raises(ValueError):
-            storage.get_flow()
-
-        @prefect.task
-        def add_to_dict():
-            with open(os.path.join(directory, "output"), "w") as tmp:
-                tmp.write("success")
-
-        flow_dir = os.path.join(directory, "flows")
-        os.makedirs(flow_dir, exist_ok=True)
-
-        with open(os.path.join(flow_dir, "test.prefect"), "w+") as env:
-            flow = Flow("test", tasks=[add_to_dict])
-            flow_path = os.path.join(flow_dir, "test.prefect")
-            with open(flow_path, "wb") as f:
-                cloudpickle.dump(flow, f)
-            out = storage.add_flow(flow)
-
-        f = storage.get_flow(out)
-        assert isinstance(f, Flow)
-        assert f.name == "test"
-        assert len(f.tasks) == 1
+    flow = Flow("test")
+    flow_path = os.path.join(flow_dir, "test.prefect")
+    with open(flow_path, "wb") as f:
+        cloudpickle.dump(flow, f)
+
+    storage = Docker(base_image="python:3.6", prefect_directory=str(tmpdir))
+    storage.add_flow(flow)
+
+    f = storage.get_flow(flow.name)
+    assert isinstance(f, Flow)
+    assert f.name == "test"
 
 
 def test_add_similar_flows_fails():
diff --git a/tests/storage/test_gcs_storage.py b/tests/storage/test_gcs_storage.py
index 6f1e99d7f1..405cc6ac12 100644
--- a/tests/storage/test_gcs_storage.py
+++ b/tests/storage/test_gcs_storage.py
@@ -1,10 +1,10 @@
+import textwrap
 from unittest.mock import MagicMock, patch, call
 
 import pytest
 
 from prefect import Flow
 from prefect.storage import GCS
-from prefect.utilities.exceptions import StorageError
 from prefect.utilities.storage import flow_to_bytes_pickle, flow_from_bytes_pickle
 
 
@@ -18,7 +18,7 @@ class TestGCSStorage:
                 "google.oauth2.service_account": MagicMock(),
             },
         ):
-            from prefect.utilities.gcp import get_storage_client
+            from prefect.utilities.gcp import get_storage_client  # noqa
 
             client_util = MagicMock()
             monkeypatch.setattr("prefect.utilities.gcp.get_storage_client", client_util)
@@ -96,35 +96,39 @@ class TestGCSStorage:
         bucket_mock = MagicMock(get_blob=MagicMock(return_value=blob_mock))
         google_client.return_value.get_bucket = MagicMock(return_value=bucket_mock)
 
-        with pytest.raises(ValueError):
-            storage = GCS(bucket="test")
-            storage.get_flow()
-
         storage = GCS(bucket="awesome-bucket", key="a-place")
         storage.add_flow(f)
 
-        fetched_flow = storage.get_flow("a-place")
+        fetched_flow = storage.get_flow(f.name)
 
         assert fetched_flow.name == f.name
         bucket_mock.get_blob.assert_called_with("a-place")
         assert blob_mock.download_as_bytes.call_count == 1
 
     def test_get_flow_from_gcs_as_file(self, google_client):
-        f = Flow("awesome-flow")
-        flow_content = """from prefect import Flow\nf=Flow('awesome-flow')"""
+        f1 = Flow("flow-1")
+        f2 = Flow("flow-2")
+        flow_content = textwrap.dedent(
+            """
+            from prefect import Flow
+            f1 = Flow('flow-1')
+            f2 = Flow('flow-2')
+            """
+        )
 
         blob_mock = MagicMock(download_as_bytes=MagicMock(return_value=flow_content))
         bucket_mock = MagicMock(get_blob=MagicMock(return_value=blob_mock))
         google_client.return_value.get_bucket = MagicMock(return_value=bucket_mock)
 
         storage = GCS(bucket="awesome-bucket", key="a-place", stored_as_script=True)
-        storage.add_flow(f)
+        storage.add_flow(f1)
+        storage.add_flow(f2)
 
-        fetched_flow = storage.get_flow("a-place")
+        assert storage.get_flow(f1.name).name == f1.name
+        assert storage.get_flow(f2.name).name == f2.name
 
-        assert fetched_flow.name == f.name
         bucket_mock.get_blob.assert_called_with("a-place")
-        assert blob_mock.download_as_bytes.call_count == 1
+        assert blob_mock.download_as_bytes.call_count == 2
 
     def test_build_no_upload_if_file_and_no_local_script_path(self, google_client):
         storage = GCS(bucket="awesome-bucket", stored_as_script=True)
diff --git a/tests/storage/test_github_storage.py b/tests/storage/test_github_storage.py
index 6c744e08ef..431cfecc48 100644
--- a/tests/storage/test_github_storage.py
+++ b/tests/storage/test_github_storage.py
@@ -76,19 +76,16 @@ def test_get_flow_github(monkeypatch):
     github = MagicMock()
     monkeypatch.setattr("prefect.utilities.git.Github", github)
 
+    extract_flow_from_file = MagicMock(return_value=f)
     monkeypatch.setattr(
-        "prefect.storage.github.extract_flow_from_file",
-        MagicMock(return_value=f),
+        "prefect.storage.github.extract_flow_from_file", extract_flow_from_file
     )
 
-    with pytest.raises(ValueError):
-        storage = GitHub(repo="test/repo")
-        storage.get_flow()
-
-    storage = GitHub(repo="test/repo", path="flow")
+    storage = GitHub(repo="test/repo", path="flow", ref="my_branch")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location, ref="my_branch")
+    new_flow = storage.get_flow(f.name)
+    assert extract_flow_from_file.call_args[1]["flow_name"] == f.name
     assert new_flow.run()
diff --git a/tests/storage/test_gitlab_storage.py b/tests/storage/test_gitlab_storage.py
index 11504dbdd1..fd7166494a 100644
--- a/tests/storage/test_gitlab_storage.py
+++ b/tests/storage/test_gitlab_storage.py
@@ -81,21 +81,17 @@ def test_get_flow_gitlab(monkeypatch):
     gitlab = MagicMock()
     monkeypatch.setattr("prefect.utilities.git.Gitlab", gitlab)
 
+    extract_flow_from_file = MagicMock(return_value=f)
     monkeypatch.setattr(
         "prefect.storage.gitlab.extract_flow_from_file",
-        MagicMock(return_value=f),
+        extract_flow_from_file,
     )
 
-    with pytest.raises(ValueError) as ex:
-        storage = GitLab(repo="test/repo")
-        storage.get_flow()
-
-    assert "No flow location provided" in str(ex.value)
-
     storage = GitLab(repo="test/repo", path="flow.py")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
+    assert extract_flow_from_file.call_args[1]["flow_name"] == f.name
     assert new_flow.run()
diff --git a/tests/storage/test_local_storage.py b/tests/storage/test_local_storage.py
index ea4173fc3b..806f63f7cd 100644
--- a/tests/storage/test_local_storage.py
+++ b/tests/storage/test_local_storage.py
@@ -1,14 +1,12 @@
 import os
 import socket
+import textwrap
 
-import cloudpickle
 import pytest
 
-import prefect
 from prefect import Flow
 from prefect.engine.results import LocalResult
 from prefect.storage import Local
-from prefect.utilities.configuration import set_temporary_config
 
 
 def test_create_local_storage():
@@ -54,7 +52,7 @@ def test_add_flow_to_storage(tmpdir):
 
     assert f.name in storage
 
-    f2 = storage.get_flow(res)
+    f2 = storage.get_flow(f.name)
     assert f2.name == "test"
 
 
@@ -82,7 +80,7 @@ def test_add_flow_file_to_storage(tmpdir):
 def test_add_flow_raises_if_name_conflict(tmpdir):
     storage = Local(directory=str(tmpdir))
     f = Flow("test")
-    res = storage.add_flow(f)
+    storage.add_flow(f)
     g = Flow("test")
     with pytest.raises(ValueError, match='name "test"'):
         storage.add_flow(g)
@@ -97,32 +95,43 @@ def test_get_env_runner_raises():
 def test_get_flow_raises_if_flow_not_present():
     s = Local()
     with pytest.raises(ValueError):
-        s.get_flow()
         s.get_flow("test")
 
 
 def test_get_flow_returns_flow(tmpdir):
     storage = Local(directory=str(tmpdir))
     f = Flow("test")
-    loc = storage.add_flow(f)
-    runner = storage.get_flow(loc)
-    assert runner == f
+    storage.add_flow(f)
+    f2 = storage.get_flow(f.name)
+    assert f2 == f
 
 
 def test_get_flow_from_file_returns_flow(tmpdir):
-    contents = """from prefect import Flow\nf=Flow('test-flow')"""
+    contents = textwrap.dedent(
+        """
+        from prefect import Flow
+        f1 = Flow('flow-1')
+        f2 = Flow('flow-2')
+        """
+    )
 
     full_path = os.path.join(tmpdir, "flow.py")
 
     with open(full_path, "w") as f:
         f.write(contents)
 
-    f = Flow("test-flow")
+    f1 = Flow("flow-1")
+    f2 = Flow("flow-2")
     storage = Local(stored_as_script=True, path=full_path)
-    storage.add_flow(f)
+    storage.add_flow(f1)
+    storage.add_flow(f2)
 
-    flow = storage.get_flow(full_path)
-    assert flow.run()
+    f1_2 = storage.get_flow(f1.name)
+    f2_2 = storage.get_flow(f2.name)
+    assert f1_2.name == f1.name
+    assert f2_2.name == f2.name
+    f1_2.run()
+    f2_2.run()
 
 
 def test_containment(tmpdir):
@@ -150,7 +159,6 @@ def test_multiple_flows_in_storage(tmpdir):
     s = Local(directory=str(tmpdir))
     f = Flow("test")
     g = Flow("other")
-    z = Flow("not")
     f_loc = s.add_flow(f)
     g_loc = s.add_flow(g)
 
@@ -158,8 +166,8 @@ def test_multiple_flows_in_storage(tmpdir):
     assert "other" in s
     assert "not" not in s
 
-    assert s.get_flow(f_loc) == f
-    assert s.get_flow(g_loc) == g
+    assert s.get_flow(f.name) == f
+    assert s.get_flow(g.name) == g
 
     assert s.flows["test"] == f_loc
     assert s.flows["other"] == g_loc
diff --git a/tests/storage/test_s3_storage.py b/tests/storage/test_s3_storage.py
index d913b98438..c5c3527c6a 100644
--- a/tests/storage/test_s3_storage.py
+++ b/tests/storage/test_s3_storage.py
@@ -1,7 +1,6 @@
 import io
 from unittest.mock import MagicMock
 
-import cloudpickle
 import pytest
 
 from prefect import context, Flow
@@ -259,13 +258,11 @@ def test_get_flow_s3(monkeypatch):
 
     storage = S3(bucket="bucket")
 
-    with pytest.raises(ValueError):
-        storage.get_flow()
-
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    assert storage.get_flow(flow_location)
+    f2 = storage.get_flow(f.name)
+    assert f2.name == f.name
     assert boto3.download_fileobj.called
     assert f.name in storage
 
@@ -281,10 +278,10 @@ def test_get_flow_s3_client_error(monkeypatch):
     storage = S3(bucket="bucket")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
     with pytest.raises(ClientError):
-        storage.get_flow(flow_location)
+        storage.get_flow(f.name)
 
     assert boto3.download_fileobj.called
 
@@ -301,9 +298,9 @@ def test_get_flow_s3_bucket_key(monkeypatch):
     storage = S3(bucket="bucket", key="key")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    assert storage.get_flow(flow_location)
+    assert storage.get_flow(f.name)
     assert boto3.download_fileobj.call_args[1]["Bucket"] == "bucket"
     assert boto3.download_fileobj.call_args[1]["Key"] == "key"
 
@@ -322,7 +319,7 @@ def test_get_flow_s3_not_in_storage(monkeypatch):
     assert f.name not in storage
 
     with pytest.raises(ValueError):
-        storage.get_flow("test/test")
+        storage.get_flow(f.name)
 
 
 def test_get_flow_s3_runs(monkeypatch):
@@ -337,9 +334,9 @@ def test_get_flow_s3_runs(monkeypatch):
     storage = S3(bucket="bucket")
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
     assert boto3.download_fileobj.called
     assert f.name in storage
 
@@ -358,18 +355,20 @@ def test_get_flow_as_file_s3_runs(monkeypatch):
 
     f = Flow("test")
 
+    extract_flow_from_file = MagicMock(return_value=f)
     monkeypatch.setattr(
         "prefect.storage.s3.extract_flow_from_file",
-        MagicMock(return_value=f),
+        extract_flow_from_file,
     )
 
     storage = S3(bucket="bucket", stored_as_script=True)
 
     assert f.name not in storage
-    flow_location = storage.add_flow(f)
+    storage.add_flow(f)
 
-    new_flow = storage.get_flow(flow_location)
+    new_flow = storage.get_flow(f.name)
     assert boto3.download_fileobj.called
+    assert extract_flow_from_file.call_args[1]["flow_name"] == f.name
     assert f.name in storage
 
     assert isinstance(new_flow, Flow)
@@ -378,54 +377,3 @@ def test_get_flow_as_file_s3_runs(monkeypatch):
 
     state = new_flow.run()
     assert state.is_successful()
-
-
-def test_get_flow_s3_from_init_key(monkeypatch):
-    client = MagicMock()
-    boto3 = MagicMock(download_fileobj=MagicMock(return_value=client))
-    monkeypatch.setattr("prefect.storage.S3._boto3_client", boto3)
-
-    f = Flow("test")
-
-    monkeypatch.setattr("cloudpickle.loads", MagicMock(return_value=f))
-
-    storage = S3(bucket="bucket", key="key")
-
-    assert len(storage.flows) == 0
-    assert len(storage._flows) == 0
-
-    downloaded_flow = storage.get_flow()
-
-    assert downloaded_flow.name == "test"
-
-
-def test_get_flow_s3_from_init_key_run(monkeypatch):
-    client = MagicMock()
-    boto3 = MagicMock(download_fileobj=MagicMock(return_value=client))
-    monkeypatch.setattr("prefect.storage.S3._boto3_client", boto3)
-
-    f = Flow("test")
-
-    monkeypatch.setattr("cloudpickle.loads", MagicMock(return_value=f))
-
-    storage = S3(bucket="bucket", key="key")
-
-    downloaded_flow = storage.get_flow()
-    assert isinstance(downloaded_flow, Flow)
-    state = downloaded_flow.run()
-    assert state.is_successful()
-
-
-def test_get_flow_s3_no_location_set(monkeypatch):
-    client = MagicMock()
-    boto3 = MagicMock(download_fileobj=MagicMock(return_value=client))
-    monkeypatch.setattr("prefect.storage.S3._boto3_client", boto3)
-
-    f = Flow("test")
-
-    monkeypatch.setattr("cloudpickle.loads", MagicMock(return_value=f))
-
-    storage = S3(bucket="bucket")
-
-    with pytest.raises(ValueError):
-        storage.get_flow()
diff --git a/tests/storage/test_webhook_storage.py b/tests/storage/test_webhook_storage.py
index 3864dbbe9e..f43b6fe15c 100644
--- a/tests/storage/test_webhook_storage.py
+++ b/tests/storage/test_webhook_storage.py
@@ -143,7 +143,7 @@ def test_webhook_build_works_with_no_arguments(sample_flow):
     res = webhook.build()
     assert isinstance(res, Webhook)
 
-    res = webhook.get_flow()
+    res = webhook.get_flow(sample_flow.name)
     assert isinstance(res, Flow)
 
 
@@ -221,7 +221,7 @@ def test_webhook_raises_error_on_get_flow_failure(sample_flow):
     webhook.build()
 
     with pytest.raises(HTTPError, match="test-error-message"):
-        webhook.get_flow()
+        webhook.get_flow(sample_flow.name)
 
 
 def test_render_dict_gets_env_variables(monkeypatch):
@@ -360,7 +360,7 @@ def test_webhook_works_with_file_storage(sample_flow, tmpdir):
     res = webhook.build()
     assert isinstance(res, Webhook)
 
-    res = webhook.get_flow()
+    res = webhook.get_flow(sample_flow.name)
     assert isinstance(res, Flow)
     assert res.name == "test-flow"