We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
document loading external files with `Git` storage
## Current behavior
<!-- Please describe how the feature works today -->
No docs


## Proposed behavior
<!-- Please describe your proposed change to the current behavior -->
Create an example doc illustration how to use `Git` storage to load external files (non-python files) from a repo with `Git` storage



## Example
<!-- Please give an example of how the enhancement would be useful -->


--- 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/utilities/storage.py`
Content:
```
1 import binascii
2 import importlib
3 import json
4 import sys
5 import warnings
6 from operator import attrgetter
7 from typing import TYPE_CHECKING
8 from distutils.version import LooseVersion
9 
10 import cloudpickle
11 
12 import prefect
13 from prefect.exceptions import FlowStorageError
14 
15 if TYPE_CHECKING:
16     from prefect.core.flow import Flow  # pylint: disable=W0611
17 
18 
19 def get_flow_image(flow: "Flow") -> str:
20     """
21     Retrieve the image to use for this flow deployment. Will start by looking for
22     an `image` value in the flow's `environment.metadata`. If not found then it will fall
23     back to using the `flow.storage`.
24 
25     Args:
26         - flow (Flow): A flow object
27 
28     Returns:
29         - str: a full image name to use for this flow run
30 
31     Raises:
32         - ValueError: if deployment attempted on unsupported Storage type and `image` not
33             present in environment metadata
34     """
35     environment = flow.environment
36     if (
37         environment is not None
38         and hasattr(environment, "metadata")
39         and environment.metadata.get("image")
40     ):
41         return environment.metadata.get("image", "")
42     else:
43         storage = flow.storage
44         if not isinstance(storage, prefect.storage.Docker):
45             raise ValueError(
46                 f"Storage for flow run {flow.name} is not of type Docker and "
47                 f"environment has no `image` attribute in the metadata field."
48             )
49 
50         return storage.name
51 
52 
53 def extract_flow_from_file(
54     file_path: str = None, file_contents: str = None, flow_name: str = None
55 ) -> "Flow":
56     """
57     Extract a flow object from a file.
58 
59     Args:
60         - file_path (str, optional): A file path pointing to a .py file containing a flow
61         - file_contents (str, optional): The string contents of a .py file containing a flow
62         - flow_name (str, optional): A specific name of a flow to extract from a file.
63             If not set then the first flow object retrieved from file will be returned.
64 
65     Returns:
66         - Flow: A flow object extracted from a file
67 
68     Raises:
69         - ValueError: if both `file_path` and `file_contents` are provided or neither are.
70     """
71     if file_path is not None:
72         if file_contents is not None:
73             raise ValueError(
74                 "Provide either `file_path` or `file_contents` but not both."
75             )
76 
77         with open(file_path, "r") as f:
78             contents = f.read()
79     elif file_contents is not None:
80         contents = file_contents
81     else:
82         raise ValueError("Provide either `file_path` or `file_contents`.")
83 
84     # Load objects from file into dict
85     exec_vals = {}  # type: ignore
86     exec(contents, exec_vals)
87 
88     # Grab flow name from values loaded via exec
89     flows = {o.name: o for o in exec_vals.values() if isinstance(o, prefect.Flow)}
90     if flows:
91         if flow_name:
92             if flow_name in flows:
93                 return flows[flow_name]
94             else:
95                 flows_list = "\n".join("- %r" % n for n in sorted(flows))
96                 raise ValueError(
97                     f"Flow {flow_name!r} not found in file. Found flows:\n{flows_list}"
98                 )
99         else:
100             return list(flows.values())[0]
101     else:
102         raise ValueError("No flows found in file.")
103 
104 
105 def extract_flow_from_module(module_str: str, flow_name: str = None) -> "Flow":
106     """
107     Extract a flow object from a python module.
108 
109     Args:
110         - module_str (str): A module path pointing to a .py file containing a flow.
111             For example, 'myrepo.mymodule.myflow' where myflow.py contains the flow.
112             Additionally, `:` can be used to access module's attribute, for example,
113             'myrepo.mymodule.myflow:flow' or 'myrepo.mymodule.myflow:MyObj.newflow'.
114         - flow_name (str, optional): A specific name of a flow to extract from a file.
115             If not provided, the `module_str` must have an attribute specifier
116             or only one `Flow` object must be present in the module, otherwise
117             an error will be raised.
118 
119     Returns:
120         - Flow: A flow object extracted from a file
121     """
122     mod_name, obj_name_present, obj_name = module_str.partition(":")
123 
124     module = importlib.import_module(mod_name)
125 
126     if obj_name_present:
127         try:
128             flow = attrgetter(obj_name)(module)
129         except AttributeError:
130             raise ValueError(f"Failed to find flow at {module_str!r}") from None
131         if callable(flow):
132             flow = flow()
133         if not isinstance(flow, prefect.Flow):
134             raise TypeError(
135                 f"Object at {module_str!r} is a {type(flow)} not a `prefect.Flow`"
136             )
137         if flow_name is not None and flow.name != flow_name:
138             raise ValueError(
139                 f"Flow at {module_str!r} is named {flow.name!r}, expected {flow_name!r}"
140             )
141         return flow
142     else:
143         flows = {}
144         for attr in dir(module):
145             obj = getattr(module, attr, None)
146             if isinstance(obj, prefect.Flow):
147                 flows[obj.name] = obj
148         if flow_name is not None:
149             if flow_name in flows:
150                 return flows[flow_name]
151             raise ValueError(f"Failed to find flow {flow_name!r} in {module_str!r}")
152         elif len(flows) == 1:
153             return flows.popitem()[1]
154         elif len(flows) > 1:
155             raise ValueError(
156                 "Multiple flows found in {module_str!r}, please provide `flow_name` to select one."
157             )
158         else:
159             raise ValueError("No flows found in {module_str!r}")
160 
161 
162 def flow_to_bytes_pickle(flow: "Flow") -> bytes:
163     """Serialize a flow to bytes.
164 
165     The flow is serialized using `cloudpickle`, with some extra metadata on
166     included via JSON. The flow can be reloaded using `flow_from_bytes_pickle`.
167 
168     Args:
169         - flow (Flow): the flow to be serialized.
170 
171     Returns:
172         - bytes: a serialized representation of the flow.
173     """
174     flow_data = binascii.b2a_base64(
175         cloudpickle.dumps(flow, protocol=4), newline=False
176     ).decode("utf-8")
177     out = json.dumps({"flow": flow_data, "versions": _get_versions()})
178     return out.encode("utf-8")
179 
180 
181 def _get_versions() -> dict:
182     """Get version info on libraries where a version-mismatch between
183     registration and execution environment may cause a flow to fail to load
184     properly"""
185     return {
186         "cloudpickle": cloudpickle.__version__,
187         "prefect": prefect.__version__,
188         "python": "%d.%d.%d" % sys.version_info[:3],
189     }
190 
191 
192 def flow_from_bytes_pickle(data: bytes) -> "Flow":
193     """Load a flow from bytes."""
194     try:
195         info = json.loads(data.decode("utf-8"))
196     except Exception:
197         # Serialized using older version of prefect, use cloudpickle directly
198         flow_bytes = data
199         reg_versions = {}
200     else:
201         flow_bytes = binascii.a2b_base64(info["flow"])
202         reg_versions = info["versions"]
203 
204     run_versions = _get_versions()
205 
206     try:
207         flow = cloudpickle.loads(flow_bytes)
208     except Exception as exc:
209         parts = ["An error occurred while unpickling the flow:", f"  {exc!r}"]
210         # Check for mismatched versions to provide a better error if possible
211         mismatches = []
212         for name, v1 in sorted(reg_versions.items()):
213             if name in run_versions:
214                 v2 = run_versions[name]
215                 if LooseVersion(v1) != v2:
216                     mismatches.append(
217                         f"  - {name}: (flow built with {v1!r}, currently running with {v2!r})"
218                     )
219         if mismatches:
220             parts.append(
221                 "This may be due to one of the following version mismatches between "
222                 "the flow build and execution environments:"
223             )
224             parts.extend(mismatches)
225         if isinstance(exc, ImportError):
226             # If it's an import error, also note that the user may need to package
227             # their dependencies
228             prefix = "This also may" if mismatches else "This may"
229             parts.append(
230                 f"{prefix} be due to a missing Python module in your current "
231                 "environment. Please ensure you have all required flow "
232                 "dependencies installed."
233             )
234         raise FlowStorageError("\n".join(parts)) from exc
235 
236     run_prefect = run_versions["prefect"]
237     reg_prefect = reg_versions.get("prefect")
238     if reg_prefect and LooseVersion(reg_prefect) != run_prefect:
239         # If we didn't error above, still check that the prefect versions match
240         # and warn if they don't. Prefect version mismatches *may* work, but
241         # they may also error later leading to confusing behavior.
242         warnings.warn(
243             f"This flow was built using Prefect {reg_prefect!r}, but you currently "
244             f"have Prefect {run_prefect!r} installed. We recommend loading flows "
245             "with the same Prefect version they were built with, failure to do so "
246             "may result in errors."
247         )
248     return flow
249
```
--- 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/utilities/storage.py b/src/prefect/utilities/storage.py
--- a/src/prefect/utilities/storage.py
+++ b/src/prefect/utilities/storage.py
@@ -4,7 +4,7 @@
 import sys
 import warnings
 from operator import attrgetter
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Dict, Any
 from distutils.version import LooseVersion
 
 import cloudpickle
@@ -82,7 +82,9 @@
         raise ValueError("Provide either `file_path` or `file_contents`.")
 
     # Load objects from file into dict
-    exec_vals = {}  # type: ignore
+    # if a file_path has been provided, provide __file__ as a global variable
+    # so it resolves correctly during extraction
+    exec_vals: Dict[str, Any] = {"__file__": file_path} if file_path else {}
     exec(contents, exec_vals)
 
     # Grab flow name from values loaded via exec

{"golden_diff": "diff --git a/src/prefect/utilities/storage.py b/src/prefect/utilities/storage.py\n--- a/src/prefect/utilities/storage.py\n+++ b/src/prefect/utilities/storage.py\n@@ -4,7 +4,7 @@\n import sys\n import warnings\n from operator import attrgetter\n-from typing import TYPE_CHECKING\n+from typing import TYPE_CHECKING, Dict, Any\n from distutils.version import LooseVersion\n \n import cloudpickle\n@@ -82,7 +82,9 @@\n         raise ValueError(\"Provide either `file_path` or `file_contents`.\")\n \n     # Load objects from file into dict\n-    exec_vals = {}  # type: ignore\n+    # if a file_path has been provided, provide __file__ as a global variable\n+    # so it resolves correctly during extraction\n+    exec_vals: Dict[str, Any] = {\"__file__\": file_path} if file_path else {}\n     exec(contents, exec_vals)\n \n     # Grab flow name from values loaded via exec\n", "issue": "document loading external files with `Git` storage\n## Current behavior\r\n<!-- Please describe how the feature works today -->\r\nNo docs\r\n\r\n\r\n## Proposed behavior\r\n<!-- Please describe your proposed change to the current behavior -->\r\nCreate an example doc illustration how to use `Git` storage to load external files (non-python files) from a repo with `Git` storage\r\n\r\n\r\n\r\n## Example\r\n<!-- Please give an example of how the enhancement would be useful -->\r\n\n", "before_files": [{"content": "import binascii\nimport importlib\nimport json\nimport sys\nimport warnings\nfrom operator import attrgetter\nfrom typing import TYPE_CHECKING\nfrom distutils.version import LooseVersion\n\nimport cloudpickle\n\nimport prefect\nfrom prefect.exceptions import FlowStorageError\n\nif TYPE_CHECKING:\n    from prefect.core.flow import Flow  # pylint: disable=W0611\n\n\ndef get_flow_image(flow: \"Flow\") -> str:\n    \"\"\"\n    Retrieve the image to use for this flow deployment. Will start by looking for\n    an `image` value in the flow's `environment.metadata`. If not found then it will fall\n    back to using the `flow.storage`.\n\n    Args:\n        - flow (Flow): A flow object\n\n    Returns:\n        - str: a full image name to use for this flow run\n\n    Raises:\n        - ValueError: if deployment attempted on unsupported Storage type and `image` not\n            present in environment metadata\n    \"\"\"\n    environment = flow.environment\n    if (\n        environment is not None\n        and hasattr(environment, \"metadata\")\n        and environment.metadata.get(\"image\")\n    ):\n        return environment.metadata.get(\"image\", \"\")\n    else:\n        storage = flow.storage\n        if not isinstance(storage, prefect.storage.Docker):\n            raise ValueError(\n                f\"Storage for flow run {flow.name} is not of type Docker and \"\n                f\"environment has no `image` attribute in the metadata field.\"\n            )\n\n        return storage.name\n\n\ndef extract_flow_from_file(\n    file_path: str = None, file_contents: str = None, flow_name: str = None\n) -> \"Flow\":\n    \"\"\"\n    Extract a flow object from a file.\n\n    Args:\n        - file_path (str, optional): A file path pointing to a .py file containing a flow\n        - file_contents (str, optional): The string contents of a .py file containing a flow\n        - flow_name (str, optional): A specific name of a flow to extract from a file.\n            If not set then the first flow object retrieved from file will be returned.\n\n    Returns:\n        - Flow: A flow object extracted from a file\n\n    Raises:\n        - ValueError: if both `file_path` and `file_contents` are provided or neither are.\n    \"\"\"\n    if file_path is not None:\n        if file_contents is not None:\n            raise ValueError(\n                \"Provide either `file_path` or `file_contents` but not both.\"\n            )\n\n        with open(file_path, \"r\") as f:\n            contents = f.read()\n    elif file_contents is not None:\n        contents = file_contents\n    else:\n        raise ValueError(\"Provide either `file_path` or `file_contents`.\")\n\n    # Load objects from file into dict\n    exec_vals = {}  # type: ignore\n    exec(contents, exec_vals)\n\n    # Grab flow name from values loaded via exec\n    flows = {o.name: o for o in exec_vals.values() if isinstance(o, prefect.Flow)}\n    if flows:\n        if flow_name:\n            if flow_name in flows:\n                return flows[flow_name]\n            else:\n                flows_list = \"\\n\".join(\"- %r\" % n for n in sorted(flows))\n                raise ValueError(\n                    f\"Flow {flow_name!r} not found in file. Found flows:\\n{flows_list}\"\n                )\n        else:\n            return list(flows.values())[0]\n    else:\n        raise ValueError(\"No flows found in file.\")\n\n\ndef extract_flow_from_module(module_str: str, flow_name: str = None) -> \"Flow\":\n    \"\"\"\n    Extract a flow object from a python module.\n\n    Args:\n        - module_str (str): A module path pointing to a .py file containing a flow.\n            For example, 'myrepo.mymodule.myflow' where myflow.py contains the flow.\n            Additionally, `:` can be used to access module's attribute, for example,\n            'myrepo.mymodule.myflow:flow' or 'myrepo.mymodule.myflow:MyObj.newflow'.\n        - flow_name (str, optional): A specific name of a flow to extract from a file.\n            If not provided, the `module_str` must have an attribute specifier\n            or only one `Flow` object must be present in the module, otherwise\n            an error will be raised.\n\n    Returns:\n        - Flow: A flow object extracted from a file\n    \"\"\"\n    mod_name, obj_name_present, obj_name = module_str.partition(\":\")\n\n    module = importlib.import_module(mod_name)\n\n    if obj_name_present:\n        try:\n            flow = attrgetter(obj_name)(module)\n        except AttributeError:\n            raise ValueError(f\"Failed to find flow at {module_str!r}\") from None\n        if callable(flow):\n            flow = flow()\n        if not isinstance(flow, prefect.Flow):\n            raise TypeError(\n                f\"Object at {module_str!r} is a {type(flow)} not a `prefect.Flow`\"\n            )\n        if flow_name is not None and flow.name != flow_name:\n            raise ValueError(\n                f\"Flow at {module_str!r} is named {flow.name!r}, expected {flow_name!r}\"\n            )\n        return flow\n    else:\n        flows = {}\n        for attr in dir(module):\n            obj = getattr(module, attr, None)\n            if isinstance(obj, prefect.Flow):\n                flows[obj.name] = obj\n        if flow_name is not None:\n            if flow_name in flows:\n                return flows[flow_name]\n            raise ValueError(f\"Failed to find flow {flow_name!r} in {module_str!r}\")\n        elif len(flows) == 1:\n            return flows.popitem()[1]\n        elif len(flows) > 1:\n            raise ValueError(\n                \"Multiple flows found in {module_str!r}, please provide `flow_name` to select one.\"\n            )\n        else:\n            raise ValueError(\"No flows found in {module_str!r}\")\n\n\ndef flow_to_bytes_pickle(flow: \"Flow\") -> bytes:\n    \"\"\"Serialize a flow to bytes.\n\n    The flow is serialized using `cloudpickle`, with some extra metadata on\n    included via JSON. The flow can be reloaded using `flow_from_bytes_pickle`.\n\n    Args:\n        - flow (Flow): the flow to be serialized.\n\n    Returns:\n        - bytes: a serialized representation of the flow.\n    \"\"\"\n    flow_data = binascii.b2a_base64(\n        cloudpickle.dumps(flow, protocol=4), newline=False\n    ).decode(\"utf-8\")\n    out = json.dumps({\"flow\": flow_data, \"versions\": _get_versions()})\n    return out.encode(\"utf-8\")\n\n\ndef _get_versions() -> dict:\n    \"\"\"Get version info on libraries where a version-mismatch between\n    registration and execution environment may cause a flow to fail to load\n    properly\"\"\"\n    return {\n        \"cloudpickle\": cloudpickle.__version__,\n        \"prefect\": prefect.__version__,\n        \"python\": \"%d.%d.%d\" % sys.version_info[:3],\n    }\n\n\ndef flow_from_bytes_pickle(data: bytes) -> \"Flow\":\n    \"\"\"Load a flow from bytes.\"\"\"\n    try:\n        info = json.loads(data.decode(\"utf-8\"))\n    except Exception:\n        # Serialized using older version of prefect, use cloudpickle directly\n        flow_bytes = data\n        reg_versions = {}\n    else:\n        flow_bytes = binascii.a2b_base64(info[\"flow\"])\n        reg_versions = info[\"versions\"]\n\n    run_versions = _get_versions()\n\n    try:\n        flow = cloudpickle.loads(flow_bytes)\n    except Exception as exc:\n        parts = [\"An error occurred while unpickling the flow:\", f\"  {exc!r}\"]\n        # Check for mismatched versions to provide a better error if possible\n        mismatches = []\n        for name, v1 in sorted(reg_versions.items()):\n            if name in run_versions:\n                v2 = run_versions[name]\n                if LooseVersion(v1) != v2:\n                    mismatches.append(\n                        f\"  - {name}: (flow built with {v1!r}, currently running with {v2!r})\"\n                    )\n        if mismatches:\n            parts.append(\n                \"This may be due to one of the following version mismatches between \"\n                \"the flow build and execution environments:\"\n            )\n            parts.extend(mismatches)\n        if isinstance(exc, ImportError):\n            # If it's an import error, also note that the user may need to package\n            # their dependencies\n            prefix = \"This also may\" if mismatches else \"This may\"\n            parts.append(\n                f\"{prefix} be due to a missing Python module in your current \"\n                \"environment. Please ensure you have all required flow \"\n                \"dependencies installed.\"\n            )\n        raise FlowStorageError(\"\\n\".join(parts)) from exc\n\n    run_prefect = run_versions[\"prefect\"]\n    reg_prefect = reg_versions.get(\"prefect\")\n    if reg_prefect and LooseVersion(reg_prefect) != run_prefect:\n        # If we didn't error above, still check that the prefect versions match\n        # and warn if they don't. Prefect version mismatches *may* work, but\n        # they may also error later leading to confusing behavior.\n        warnings.warn(\n            f\"This flow was built using Prefect {reg_prefect!r}, but you currently \"\n            f\"have Prefect {run_prefect!r} installed. We recommend loading flows \"\n            \"with the same Prefect version they were built with, failure to do so \"\n            \"may result in errors.\"\n        )\n    return flow\n", "path": "src/prefect/utilities/storage.py"}], "after_files": [{"content": "import binascii\nimport importlib\nimport json\nimport sys\nimport warnings\nfrom operator import attrgetter\nfrom typing import TYPE_CHECKING, Dict, Any\nfrom distutils.version import LooseVersion\n\nimport cloudpickle\n\nimport prefect\nfrom prefect.exceptions import FlowStorageError\n\nif TYPE_CHECKING:\n    from prefect.core.flow import Flow  # pylint: disable=W0611\n\n\ndef get_flow_image(flow: \"Flow\") -> str:\n    \"\"\"\n    Retrieve the image to use for this flow deployment. Will start by looking for\n    an `image` value in the flow's `environment.metadata`. If not found then it will fall\n    back to using the `flow.storage`.\n\n    Args:\n        - flow (Flow): A flow object\n\n    Returns:\n        - str: a full image name to use for this flow run\n\n    Raises:\n        - ValueError: if deployment attempted on unsupported Storage type and `image` not\n            present in environment metadata\n    \"\"\"\n    environment = flow.environment\n    if (\n        environment is not None\n        and hasattr(environment, \"metadata\")\n        and environment.metadata.get(\"image\")\n    ):\n        return environment.metadata.get(\"image\", \"\")\n    else:\n        storage = flow.storage\n        if not isinstance(storage, prefect.storage.Docker):\n            raise ValueError(\n                f\"Storage for flow run {flow.name} is not of type Docker and \"\n                f\"environment has no `image` attribute in the metadata field.\"\n            )\n\n        return storage.name\n\n\ndef extract_flow_from_file(\n    file_path: str = None, file_contents: str = None, flow_name: str = None\n) -> \"Flow\":\n    \"\"\"\n    Extract a flow object from a file.\n\n    Args:\n        - file_path (str, optional): A file path pointing to a .py file containing a flow\n        - file_contents (str, optional): The string contents of a .py file containing a flow\n        - flow_name (str, optional): A specific name of a flow to extract from a file.\n            If not set then the first flow object retrieved from file will be returned.\n\n    Returns:\n        - Flow: A flow object extracted from a file\n\n    Raises:\n        - ValueError: if both `file_path` and `file_contents` are provided or neither are.\n    \"\"\"\n    if file_path is not None:\n        if file_contents is not None:\n            raise ValueError(\n                \"Provide either `file_path` or `file_contents` but not both.\"\n            )\n\n        with open(file_path, \"r\") as f:\n            contents = f.read()\n    elif file_contents is not None:\n        contents = file_contents\n    else:\n        raise ValueError(\"Provide either `file_path` or `file_contents`.\")\n\n    # Load objects from file into dict\n    # if a file_path has been provided, provide __file__ as a global variable\n    # so it resolves correctly during extraction\n    exec_vals: Dict[str, Any] = {\"__file__\": file_path} if file_path else {}\n    exec(contents, exec_vals)\n\n    # Grab flow name from values loaded via exec\n    flows = {o.name: o for o in exec_vals.values() if isinstance(o, prefect.Flow)}\n    if flows:\n        if flow_name:\n            if flow_name in flows:\n                return flows[flow_name]\n            else:\n                flows_list = \"\\n\".join(\"- %r\" % n for n in sorted(flows))\n                raise ValueError(\n                    f\"Flow {flow_name!r} not found in file. Found flows:\\n{flows_list}\"\n                )\n        else:\n            return list(flows.values())[0]\n    else:\n        raise ValueError(\"No flows found in file.\")\n\n\ndef extract_flow_from_module(module_str: str, flow_name: str = None) -> \"Flow\":\n    \"\"\"\n    Extract a flow object from a python module.\n\n    Args:\n        - module_str (str): A module path pointing to a .py file containing a flow.\n            For example, 'myrepo.mymodule.myflow' where myflow.py contains the flow.\n            Additionally, `:` can be used to access module's attribute, for example,\n            'myrepo.mymodule.myflow:flow' or 'myrepo.mymodule.myflow:MyObj.newflow'.\n        - flow_name (str, optional): A specific name of a flow to extract from a file.\n            If not provided, the `module_str` must have an attribute specifier\n            or only one `Flow` object must be present in the module, otherwise\n            an error will be raised.\n\n    Returns:\n        - Flow: A flow object extracted from a file\n    \"\"\"\n    mod_name, obj_name_present, obj_name = module_str.partition(\":\")\n\n    module = importlib.import_module(mod_name)\n\n    if obj_name_present:\n        try:\n            flow = attrgetter(obj_name)(module)\n        except AttributeError:\n            raise ValueError(f\"Failed to find flow at {module_str!r}\") from None\n        if callable(flow):\n            flow = flow()\n        if not isinstance(flow, prefect.Flow):\n            raise TypeError(\n                f\"Object at {module_str!r} is a {type(flow)} not a `prefect.Flow`\"\n            )\n        if flow_name is not None and flow.name != flow_name:\n            raise ValueError(\n                f\"Flow at {module_str!r} is named {flow.name!r}, expected {flow_name!r}\"\n            )\n        return flow\n    else:\n        flows = {}\n        for attr in dir(module):\n            obj = getattr(module, attr, None)\n            if isinstance(obj, prefect.Flow):\n                flows[obj.name] = obj\n        if flow_name is not None:\n            if flow_name in flows:\n                return flows[flow_name]\n            raise ValueError(f\"Failed to find flow {flow_name!r} in {module_str!r}\")\n        elif len(flows) == 1:\n            return flows.popitem()[1]\n        elif len(flows) > 1:\n            raise ValueError(\n                \"Multiple flows found in {module_str!r}, please provide `flow_name` to select one.\"\n            )\n        else:\n            raise ValueError(\"No flows found in {module_str!r}\")\n\n\ndef flow_to_bytes_pickle(flow: \"Flow\") -> bytes:\n    \"\"\"Serialize a flow to bytes.\n\n    The flow is serialized using `cloudpickle`, with some extra metadata on\n    included via JSON. The flow can be reloaded using `flow_from_bytes_pickle`.\n\n    Args:\n        - flow (Flow): the flow to be serialized.\n\n    Returns:\n        - bytes: a serialized representation of the flow.\n    \"\"\"\n    flow_data = binascii.b2a_base64(\n        cloudpickle.dumps(flow, protocol=4), newline=False\n    ).decode(\"utf-8\")\n    out = json.dumps({\"flow\": flow_data, \"versions\": _get_versions()})\n    return out.encode(\"utf-8\")\n\n\ndef _get_versions() -> dict:\n    \"\"\"Get version info on libraries where a version-mismatch between\n    registration and execution environment may cause a flow to fail to load\n    properly\"\"\"\n    return {\n        \"cloudpickle\": cloudpickle.__version__,\n        \"prefect\": prefect.__version__,\n        \"python\": \"%d.%d.%d\" % sys.version_info[:3],\n    }\n\n\ndef flow_from_bytes_pickle(data: bytes) -> \"Flow\":\n    \"\"\"Load a flow from bytes.\"\"\"\n    try:\n        info = json.loads(data.decode(\"utf-8\"))\n    except Exception:\n        # Serialized using older version of prefect, use cloudpickle directly\n        flow_bytes = data\n        reg_versions = {}\n    else:\n        flow_bytes = binascii.a2b_base64(info[\"flow\"])\n        reg_versions = info[\"versions\"]\n\n    run_versions = _get_versions()\n\n    try:\n        flow = cloudpickle.loads(flow_bytes)\n    except Exception as exc:\n        parts = [\"An error occurred while unpickling the flow:\", f\"  {exc!r}\"]\n        # Check for mismatched versions to provide a better error if possible\n        mismatches = []\n        for name, v1 in sorted(reg_versions.items()):\n            if name in run_versions:\n                v2 = run_versions[name]\n                if LooseVersion(v1) != v2:\n                    mismatches.append(\n                        f\"  - {name}: (flow built with {v1!r}, currently running with {v2!r})\"\n                    )\n        if mismatches:\n            parts.append(\n                \"This may be due to one of the following version mismatches between \"\n                \"the flow build and execution environments:\"\n            )\n            parts.extend(mismatches)\n        if isinstance(exc, ImportError):\n            # If it's an import error, also note that the user may need to package\n            # their dependencies\n            prefix = \"This also may\" if mismatches else \"This may\"\n            parts.append(\n                f\"{prefix} be due to a missing Python module in your current \"\n                \"environment. Please ensure you have all required flow \"\n                \"dependencies installed.\"\n            )\n        raise FlowStorageError(\"\\n\".join(parts)) from exc\n\n    run_prefect = run_versions[\"prefect\"]\n    reg_prefect = reg_versions.get(\"prefect\")\n    if reg_prefect and LooseVersion(reg_prefect) != run_prefect:\n        # If we didn't error above, still check that the prefect versions match\n        # and warn if they don't. Prefect version mismatches *may* work, but\n        # they may also error later leading to confusing behavior.\n        warnings.warn(\n            f\"This flow was built using Prefect {reg_prefect!r}, but you currently \"\n            f\"have Prefect {run_prefect!r} installed. We recommend loading flows \"\n            \"with the same Prefect version they were built with, failure to do so \"\n            \"may result in errors.\"\n        )\n    return flow\n", "path": "src/prefect/utilities/storage.py"}]}