We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Windows prefect deploy does not work with docker
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

prefect deploy -n "name of deployment" does not work anymore on Windows. 

### Reproduction

```python3
prefect deploy -n
```


### Error

```python3
Running deployment build steps...
 > Running run_shell_script step...
 > Running build_docker_image step...
Unable to load step function: prefect_docker.deployments.steps.build_docker_image. Attempting install of prefect-docker>=0.3.11.
Traceback (most recent call last):
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\deployments\steps\core.py", line 149, in run_steps
    step_output = await run_step(step, upstream_outputs)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\deployments\steps\core.py", line 119, in run_step
    step_func = _get_function_for_step(fqn, requires=keywords.get("requires"))
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\deployments\steps\core.py", line 83, in _get_function_for_step
    subprocess.check_call(
  File "C:\Users\JKANG1\AppData\Local\Programs\Python\Python311\Lib\subprocess.py", line 408, in check_call
    retcode = call(*popenargs, **kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\AppData\Local\Programs\Python\Python311\Lib\subprocess.py", line 389, in call
    with Popen(*popenargs, **kwargs) as p:
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\AppData\Local\Programs\Python\Python311\Lib\subprocess.py", line 1026, in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
  File "C:\Users\JKANG1\AppData\Local\Programs\Python\Python311\Lib\subprocess.py", line 1538, in _execute_child
    hp, ht, pid, tid = _winapi.CreateProcess(executable, args,
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
FileNotFoundError: [WinError 2] The system cannot find the file specified

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\cli\_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\utilities\asyncutils.py", line 255, in coroutine_wrapper
    return call()
           ^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\_internal\concurrency\calls.py", line 382, in __call__
    return self.result()
           ^^^^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\_internal\concurrency\calls.py", line 282, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\_internal\concurrency\calls.py", line 168, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\AppData\Local\Programs\Python\Python311\Lib\concurrent\futures\_base.py", line 401, in __get_result
    raise self._exception
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\_internal\concurrency\calls.py", line 345, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\cli\deploy.py", line 292, in deploy
    await _run_single_deploy(
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\client\utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\cli\deploy.py", line 595, in _run_single_deploy
    await run_steps(build_steps, step_outputs, print_function=app.console.print)
  File "C:\Users\JKANG1\PycharmProjects\site_analytics\.venv\Lib\site-packages\prefect\deployments\steps\core.py", line 177, in run_steps
    raise StepExecutionError(f"Encountered error while running {fqn}") from exc
prefect.deployments.steps.core.StepExecutionError: Encountered error while running prefect_docker.deployments.steps.build_docker_image
An exception occurred.
```


### Versions

```Text
Version:             2.14.4
API version:         0.8.4
Python version:      3.11.6
Git commit:          d2cf30f4
Built:               Thu, Nov 9, 2023 4:48 PM
OS/Arch:             win32/AMD64
Profile:             default
Server type:         cloud
```


### Additional context

_No response_

--- 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/deployments/steps/core.py`
Content:
```
1 """
2 Core primitives for running Prefect project steps.
3 
4 Project steps are YAML representations of Python functions along with their inputs.
5 
6 Whenever a step is run, the following actions are taken:
7 
8 - The step's inputs and block / variable references are resolved (see [the projects concepts documentation](/concepts/projects/#templating-options) for more details)
9 - The step's function is imported; if it cannot be found, the `requires` keyword is used to install the necessary packages
10 - The step's function is called with the resolved inputs
11 - The step's output is returned and used to resolve inputs for subsequent steps
12 """
13 import os
14 import re
15 import subprocess
16 import warnings
17 from copy import deepcopy
18 from importlib import import_module
19 from typing import Any, Dict, List, Optional, Tuple, Union
20 
21 from prefect._internal.compatibility.deprecated import PrefectDeprecationWarning
22 from prefect._internal.concurrency.api import Call, from_async
23 from prefect.logging.loggers import get_logger
24 from prefect.settings import PREFECT_DEBUG_MODE
25 from prefect.utilities.importtools import import_object
26 from prefect.utilities.processutils import get_sys_executable
27 from prefect.utilities.templating import (
28     apply_values,
29     resolve_block_document_references,
30     resolve_variables,
31 )
32 
33 RESERVED_KEYWORDS = {"id", "requires"}
34 
35 
36 class StepExecutionError(Exception):
37     """
38     Raised when a step fails to execute.
39     """
40 
41 
42 def _strip_version(requirement: str) -> str:
43     """
44     Strips the version from a requirement string.
45 
46     Args:
47         requirement: A requirement string, e.g. "requests>=2.0.0"
48 
49     Returns:
50         The package name, e.g. "requests"
51 
52     Examples:
53         ```python
54         >>> _strip_version("s3fs>=2.0.0<3.0.0")
55         "s3fs"
56         ```
57     """
58     # split on any of the characters in the set [<>=!~]
59     # and return the first element which will be the package name
60     return re.split(r"[<>=!~]", requirement)[0].strip()
61 
62 
63 def _get_function_for_step(
64     fully_qualified_name: str, requires: Union[str, List[str], None] = None
65 ):
66     if not isinstance(requires, list):
67         packages = [requires] if requires else []
68     else:
69         packages = requires
70 
71     try:
72         for package in packages:
73             import_module(_strip_version(package).replace("-", "_"))
74         step_func = import_object(fully_qualified_name)
75         return step_func
76     except ImportError:
77         if requires:
78             print(
79                 f"Unable to load step function: {fully_qualified_name}. Attempting"
80                 f" install of {requires}."
81             )
82         else:
83             raise
84 
85     try:
86         subprocess.check_call(
87             [get_sys_executable(), "-m", "pip", "install", *packages],
88         )
89     except subprocess.CalledProcessError:
90         get_logger("deployments.steps.core").warning(
91             "Unable to install required packages for %s", fully_qualified_name
92         )
93     step_func = import_object(fully_qualified_name)
94     return step_func
95 
96 
97 async def run_step(step: Dict, upstream_outputs: Optional[Dict] = None) -> Dict:
98     """
99     Runs a step, returns the step's output.
100 
101     Steps are assumed to be in the format `{"importable.func.name": {"kwarg1": "value1", ...}}`.
102 
103     The 'id and 'requires' keywords are reserved for specific purposes and will be removed from the
104     inputs before passing to the step function:
105 
106     This keyword is used to specify packages that should be installed before running the step.
107     """
108     fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)
109     upstream_outputs = upstream_outputs or {}
110 
111     if len(step.keys()) > 1:
112         raise ValueError(
113             f"Step has unexpected additional keys: {', '.join(list(step.keys())[1:])}"
114         )
115 
116     keywords = {
117         keyword: inputs.pop(keyword)
118         for keyword in RESERVED_KEYWORDS
119         if keyword in inputs
120     }
121 
122     inputs = apply_values(inputs, upstream_outputs)
123     inputs = await resolve_block_document_references(inputs)
124     inputs = await resolve_variables(inputs)
125     inputs = apply_values(inputs, os.environ)
126     step_func = _get_function_for_step(fqn, requires=keywords.get("requires"))
127     result = await from_async.call_soon_in_new_thread(
128         Call.new(step_func, **inputs)
129     ).aresult()
130     return result
131 
132 
133 async def run_steps(
134     steps: List[Dict[str, Any]],
135     upstream_outputs: Optional[Dict[str, Any]] = None,
136     print_function: Any = print,
137 ):
138     upstream_outputs = deepcopy(upstream_outputs) if upstream_outputs else {}
139     for step in steps:
140         if not step:
141             continue
142         fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)
143         step_name = fqn.split(".")[-1]
144         print_function(f" > Running {step_name} step...")
145         try:
146             # catch warnings to ensure deprecation warnings are printed
147             with warnings.catch_warnings(record=True) as w:
148                 warnings.simplefilter(
149                     "always",
150                     category=PrefectDeprecationWarning,
151                 )
152                 warnings.simplefilter(
153                     "always",
154                     category=DeprecationWarning,
155                 )
156                 step_output = await run_step(step, upstream_outputs)
157             if w:
158                 printed_messages = []
159                 for warning in w:
160                     message = str(warning.message)
161                     # prevent duplicate warnings from being printed
162                     if message not in printed_messages:
163                         try:
164                             # try using rich styling
165                             print_function(message, style="yellow")
166                         except Exception:
167                             # default to printing without styling
168                             print_function(message)
169                         printed_messages.append(message)
170 
171             if not isinstance(step_output, dict):
172                 if PREFECT_DEBUG_MODE:
173                     get_logger().warning(
174                         "Step function %s returned unexpected type: %s",
175                         fqn,
176                         type(step_output),
177                     )
178                 continue
179             # store step output under step id to prevent clobbering
180             if inputs.get("id"):
181                 upstream_outputs[inputs.get("id")] = step_output
182             upstream_outputs.update(step_output)
183         except Exception as exc:
184             raise StepExecutionError(f"Encountered error while running {fqn}") from exc
185     return upstream_outputs
186 
187 
188 def _get_step_fully_qualified_name_and_inputs(step: Dict) -> Tuple[str, Dict]:
189     step = deepcopy(step)
190     return step.popitem()
191
```
--- 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/deployments/steps/core.py b/src/prefect/deployments/steps/core.py
--- a/src/prefect/deployments/steps/core.py
+++ b/src/prefect/deployments/steps/core.py
@@ -13,6 +13,7 @@
 import os
 import re
 import subprocess
+import sys
 import warnings
 from copy import deepcopy
 from importlib import import_module
@@ -23,7 +24,6 @@
 from prefect.logging.loggers import get_logger
 from prefect.settings import PREFECT_DEBUG_MODE
 from prefect.utilities.importtools import import_object
-from prefect.utilities.processutils import get_sys_executable
 from prefect.utilities.templating import (
     apply_values,
     resolve_block_document_references,
@@ -83,9 +83,7 @@
             raise
 
     try:
-        subprocess.check_call(
-            [get_sys_executable(), "-m", "pip", "install", *packages],
-        )
+        subprocess.check_call([sys.executable, "-m", "pip", "install", *packages])
     except subprocess.CalledProcessError:
         get_logger("deployments.steps.core").warning(
             "Unable to install required packages for %s", fully_qualified_name

{"golden_diff": "diff --git a/src/prefect/deployments/steps/core.py b/src/prefect/deployments/steps/core.py\n--- a/src/prefect/deployments/steps/core.py\n+++ b/src/prefect/deployments/steps/core.py\n@@ -13,6 +13,7 @@\n import os\n import re\n import subprocess\n+import sys\n import warnings\n from copy import deepcopy\n from importlib import import_module\n@@ -23,7 +24,6 @@\n from prefect.logging.loggers import get_logger\n from prefect.settings import PREFECT_DEBUG_MODE\n from prefect.utilities.importtools import import_object\n-from prefect.utilities.processutils import get_sys_executable\n from prefect.utilities.templating import (\n     apply_values,\n     resolve_block_document_references,\n@@ -83,9 +83,7 @@\n             raise\n \n     try:\n-        subprocess.check_call(\n-            [get_sys_executable(), \"-m\", \"pip\", \"install\", *packages],\n-        )\n+        subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", *packages])\n     except subprocess.CalledProcessError:\n         get_logger(\"deployments.steps.core\").warning(\n             \"Unable to install required packages for %s\", fully_qualified_name\n", "issue": "Windows prefect deploy does not work with docker\n### First check\n\n- [X] I added a descriptive title to this issue.\n- [X] I used the GitHub search to find a similar issue and didn't find it.\n- [X] I searched the Prefect documentation for this issue.\n- [X] I checked that this issue is related to Prefect and not one of its dependencies.\n\n### Bug summary\n\nprefect deploy -n \"name of deployment\" does not work anymore on Windows. \n\n### Reproduction\n\n```python3\nprefect deploy -n\n```\n\n\n### Error\n\n```python3\nRunning deployment build steps...\r\n > Running run_shell_script step...\r\n > Running build_docker_image step...\r\nUnable to load step function: prefect_docker.deployments.steps.build_docker_image. Attempting install of prefect-docker>=0.3.11.\r\nTraceback (most recent call last):\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\deployments\\steps\\core.py\", line 149, in run_steps\r\n    step_output = await run_step(step, upstream_outputs)\r\n                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\deployments\\steps\\core.py\", line 119, in run_step\r\n    step_func = _get_function_for_step(fqn, requires=keywords.get(\"requires\"))\r\n                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\deployments\\steps\\core.py\", line 83, in _get_function_for_step\r\n    subprocess.check_call(\r\n  File \"C:\\Users\\JKANG1\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\subprocess.py\", line 408, in check_call\r\n    retcode = call(*popenargs, **kwargs)\r\n              ^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\subprocess.py\", line 389, in call\r\n    with Popen(*popenargs, **kwargs) as p:\r\n         ^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\subprocess.py\", line 1026, in __init__\r\n    self._execute_child(args, executable, preexec_fn, close_fds,\r\n  File \"C:\\Users\\JKANG1\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\subprocess.py\", line 1538, in _execute_child\r\n    hp, ht, pid, tid = _winapi.CreateProcess(executable, args,\r\n                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\nFileNotFoundError: [WinError 2] The system cannot find the file specified\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nTraceback (most recent call last):\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\cli\\_utilities.py\", line 41, in wrapper\r\n    return fn(*args, **kwargs)\r\n           ^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\utilities\\asyncutils.py\", line 255, in coroutine_wrapper\r\n    return call()\r\n           ^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\_internal\\concurrency\\calls.py\", line 382, in __call__\r\n    return self.result()\r\n           ^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\_internal\\concurrency\\calls.py\", line 282, in result\r\n    return self.future.result(timeout=timeout)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\_internal\\concurrency\\calls.py\", line 168, in result\r\n    return self.__get_result()\r\n           ^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\concurrent\\futures\\_base.py\", line 401, in __get_result\r\n    raise self._exception\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\_internal\\concurrency\\calls.py\", line 345, in _run_async\r\n    result = await coro\r\n             ^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\cli\\deploy.py\", line 292, in deploy\r\n    await _run_single_deploy(\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\client\\utilities.py\", line 51, in with_injected_client\r\n    return await fn(*args, **kwargs)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\cli\\deploy.py\", line 595, in _run_single_deploy\r\n    await run_steps(build_steps, step_outputs, print_function=app.console.print)\r\n  File \"C:\\Users\\JKANG1\\PycharmProjects\\site_analytics\\.venv\\Lib\\site-packages\\prefect\\deployments\\steps\\core.py\", line 177, in run_steps\r\n    raise StepExecutionError(f\"Encountered error while running {fqn}\") from exc\r\nprefect.deployments.steps.core.StepExecutionError: Encountered error while running prefect_docker.deployments.steps.build_docker_image\r\nAn exception occurred.\n```\n\n\n### Versions\n\n```Text\nVersion:             2.14.4\r\nAPI version:         0.8.4\r\nPython version:      3.11.6\r\nGit commit:          d2cf30f4\r\nBuilt:               Thu, Nov 9, 2023 4:48 PM\r\nOS/Arch:             win32/AMD64\r\nProfile:             default\r\nServer type:         cloud\n```\n\n\n### Additional context\n\n_No response_\n", "before_files": [{"content": "\"\"\"\nCore primitives for running Prefect project steps.\n\nProject steps are YAML representations of Python functions along with their inputs.\n\nWhenever a step is run, the following actions are taken:\n\n- The step's inputs and block / variable references are resolved (see [the projects concepts documentation](/concepts/projects/#templating-options) for more details)\n- The step's function is imported; if it cannot be found, the `requires` keyword is used to install the necessary packages\n- The step's function is called with the resolved inputs\n- The step's output is returned and used to resolve inputs for subsequent steps\n\"\"\"\nimport os\nimport re\nimport subprocess\nimport warnings\nfrom copy import deepcopy\nfrom importlib import import_module\nfrom typing import Any, Dict, List, Optional, Tuple, Union\n\nfrom prefect._internal.compatibility.deprecated import PrefectDeprecationWarning\nfrom prefect._internal.concurrency.api import Call, from_async\nfrom prefect.logging.loggers import get_logger\nfrom prefect.settings import PREFECT_DEBUG_MODE\nfrom prefect.utilities.importtools import import_object\nfrom prefect.utilities.processutils import get_sys_executable\nfrom prefect.utilities.templating import (\n    apply_values,\n    resolve_block_document_references,\n    resolve_variables,\n)\n\nRESERVED_KEYWORDS = {\"id\", \"requires\"}\n\n\nclass StepExecutionError(Exception):\n    \"\"\"\n    Raised when a step fails to execute.\n    \"\"\"\n\n\ndef _strip_version(requirement: str) -> str:\n    \"\"\"\n    Strips the version from a requirement string.\n\n    Args:\n        requirement: A requirement string, e.g. \"requests>=2.0.0\"\n\n    Returns:\n        The package name, e.g. \"requests\"\n\n    Examples:\n        ```python\n        >>> _strip_version(\"s3fs>=2.0.0<3.0.0\")\n        \"s3fs\"\n        ```\n    \"\"\"\n    # split on any of the characters in the set [<>=!~]\n    # and return the first element which will be the package name\n    return re.split(r\"[<>=!~]\", requirement)[0].strip()\n\n\ndef _get_function_for_step(\n    fully_qualified_name: str, requires: Union[str, List[str], None] = None\n):\n    if not isinstance(requires, list):\n        packages = [requires] if requires else []\n    else:\n        packages = requires\n\n    try:\n        for package in packages:\n            import_module(_strip_version(package).replace(\"-\", \"_\"))\n        step_func = import_object(fully_qualified_name)\n        return step_func\n    except ImportError:\n        if requires:\n            print(\n                f\"Unable to load step function: {fully_qualified_name}. Attempting\"\n                f\" install of {requires}.\"\n            )\n        else:\n            raise\n\n    try:\n        subprocess.check_call(\n            [get_sys_executable(), \"-m\", \"pip\", \"install\", *packages],\n        )\n    except subprocess.CalledProcessError:\n        get_logger(\"deployments.steps.core\").warning(\n            \"Unable to install required packages for %s\", fully_qualified_name\n        )\n    step_func = import_object(fully_qualified_name)\n    return step_func\n\n\nasync def run_step(step: Dict, upstream_outputs: Optional[Dict] = None) -> Dict:\n    \"\"\"\n    Runs a step, returns the step's output.\n\n    Steps are assumed to be in the format `{\"importable.func.name\": {\"kwarg1\": \"value1\", ...}}`.\n\n    The 'id and 'requires' keywords are reserved for specific purposes and will be removed from the\n    inputs before passing to the step function:\n\n    This keyword is used to specify packages that should be installed before running the step.\n    \"\"\"\n    fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)\n    upstream_outputs = upstream_outputs or {}\n\n    if len(step.keys()) > 1:\n        raise ValueError(\n            f\"Step has unexpected additional keys: {', '.join(list(step.keys())[1:])}\"\n        )\n\n    keywords = {\n        keyword: inputs.pop(keyword)\n        for keyword in RESERVED_KEYWORDS\n        if keyword in inputs\n    }\n\n    inputs = apply_values(inputs, upstream_outputs)\n    inputs = await resolve_block_document_references(inputs)\n    inputs = await resolve_variables(inputs)\n    inputs = apply_values(inputs, os.environ)\n    step_func = _get_function_for_step(fqn, requires=keywords.get(\"requires\"))\n    result = await from_async.call_soon_in_new_thread(\n        Call.new(step_func, **inputs)\n    ).aresult()\n    return result\n\n\nasync def run_steps(\n    steps: List[Dict[str, Any]],\n    upstream_outputs: Optional[Dict[str, Any]] = None,\n    print_function: Any = print,\n):\n    upstream_outputs = deepcopy(upstream_outputs) if upstream_outputs else {}\n    for step in steps:\n        if not step:\n            continue\n        fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)\n        step_name = fqn.split(\".\")[-1]\n        print_function(f\" > Running {step_name} step...\")\n        try:\n            # catch warnings to ensure deprecation warnings are printed\n            with warnings.catch_warnings(record=True) as w:\n                warnings.simplefilter(\n                    \"always\",\n                    category=PrefectDeprecationWarning,\n                )\n                warnings.simplefilter(\n                    \"always\",\n                    category=DeprecationWarning,\n                )\n                step_output = await run_step(step, upstream_outputs)\n            if w:\n                printed_messages = []\n                for warning in w:\n                    message = str(warning.message)\n                    # prevent duplicate warnings from being printed\n                    if message not in printed_messages:\n                        try:\n                            # try using rich styling\n                            print_function(message, style=\"yellow\")\n                        except Exception:\n                            # default to printing without styling\n                            print_function(message)\n                        printed_messages.append(message)\n\n            if not isinstance(step_output, dict):\n                if PREFECT_DEBUG_MODE:\n                    get_logger().warning(\n                        \"Step function %s returned unexpected type: %s\",\n                        fqn,\n                        type(step_output),\n                    )\n                continue\n            # store step output under step id to prevent clobbering\n            if inputs.get(\"id\"):\n                upstream_outputs[inputs.get(\"id\")] = step_output\n            upstream_outputs.update(step_output)\n        except Exception as exc:\n            raise StepExecutionError(f\"Encountered error while running {fqn}\") from exc\n    return upstream_outputs\n\n\ndef _get_step_fully_qualified_name_and_inputs(step: Dict) -> Tuple[str, Dict]:\n    step = deepcopy(step)\n    return step.popitem()\n", "path": "src/prefect/deployments/steps/core.py"}], "after_files": [{"content": "\"\"\"\nCore primitives for running Prefect project steps.\n\nProject steps are YAML representations of Python functions along with their inputs.\n\nWhenever a step is run, the following actions are taken:\n\n- The step's inputs and block / variable references are resolved (see [the projects concepts documentation](/concepts/projects/#templating-options) for more details)\n- The step's function is imported; if it cannot be found, the `requires` keyword is used to install the necessary packages\n- The step's function is called with the resolved inputs\n- The step's output is returned and used to resolve inputs for subsequent steps\n\"\"\"\nimport os\nimport re\nimport subprocess\nimport sys\nimport warnings\nfrom copy import deepcopy\nfrom importlib import import_module\nfrom typing import Any, Dict, List, Optional, Tuple, Union\n\nfrom prefect._internal.compatibility.deprecated import PrefectDeprecationWarning\nfrom prefect._internal.concurrency.api import Call, from_async\nfrom prefect.logging.loggers import get_logger\nfrom prefect.settings import PREFECT_DEBUG_MODE\nfrom prefect.utilities.importtools import import_object\nfrom prefect.utilities.templating import (\n    apply_values,\n    resolve_block_document_references,\n    resolve_variables,\n)\n\nRESERVED_KEYWORDS = {\"id\", \"requires\"}\n\n\nclass StepExecutionError(Exception):\n    \"\"\"\n    Raised when a step fails to execute.\n    \"\"\"\n\n\ndef _strip_version(requirement: str) -> str:\n    \"\"\"\n    Strips the version from a requirement string.\n\n    Args:\n        requirement: A requirement string, e.g. \"requests>=2.0.0\"\n\n    Returns:\n        The package name, e.g. \"requests\"\n\n    Examples:\n        ```python\n        >>> _strip_version(\"s3fs>=2.0.0<3.0.0\")\n        \"s3fs\"\n        ```\n    \"\"\"\n    # split on any of the characters in the set [<>=!~]\n    # and return the first element which will be the package name\n    return re.split(r\"[<>=!~]\", requirement)[0].strip()\n\n\ndef _get_function_for_step(\n    fully_qualified_name: str, requires: Union[str, List[str], None] = None\n):\n    if not isinstance(requires, list):\n        packages = [requires] if requires else []\n    else:\n        packages = requires\n\n    try:\n        for package in packages:\n            import_module(_strip_version(package).replace(\"-\", \"_\"))\n        step_func = import_object(fully_qualified_name)\n        return step_func\n    except ImportError:\n        if requires:\n            print(\n                f\"Unable to load step function: {fully_qualified_name}. Attempting\"\n                f\" install of {requires}.\"\n            )\n        else:\n            raise\n\n    try:\n        subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", *packages])\n    except subprocess.CalledProcessError:\n        get_logger(\"deployments.steps.core\").warning(\n            \"Unable to install required packages for %s\", fully_qualified_name\n        )\n    step_func = import_object(fully_qualified_name)\n    return step_func\n\n\nasync def run_step(step: Dict, upstream_outputs: Optional[Dict] = None) -> Dict:\n    \"\"\"\n    Runs a step, returns the step's output.\n\n    Steps are assumed to be in the format `{\"importable.func.name\": {\"kwarg1\": \"value1\", ...}}`.\n\n    The 'id and 'requires' keywords are reserved for specific purposes and will be removed from the\n    inputs before passing to the step function:\n\n    This keyword is used to specify packages that should be installed before running the step.\n    \"\"\"\n    fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)\n    upstream_outputs = upstream_outputs or {}\n\n    if len(step.keys()) > 1:\n        raise ValueError(\n            f\"Step has unexpected additional keys: {', '.join(list(step.keys())[1:])}\"\n        )\n\n    keywords = {\n        keyword: inputs.pop(keyword)\n        for keyword in RESERVED_KEYWORDS\n        if keyword in inputs\n    }\n\n    inputs = apply_values(inputs, upstream_outputs)\n    inputs = await resolve_block_document_references(inputs)\n    inputs = await resolve_variables(inputs)\n    inputs = apply_values(inputs, os.environ)\n    step_func = _get_function_for_step(fqn, requires=keywords.get(\"requires\"))\n    result = await from_async.call_soon_in_new_thread(\n        Call.new(step_func, **inputs)\n    ).aresult()\n    return result\n\n\nasync def run_steps(\n    steps: List[Dict[str, Any]],\n    upstream_outputs: Optional[Dict[str, Any]] = None,\n    print_function: Any = print,\n):\n    upstream_outputs = deepcopy(upstream_outputs) if upstream_outputs else {}\n    for step in steps:\n        if not step:\n            continue\n        fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)\n        step_name = fqn.split(\".\")[-1]\n        print_function(f\" > Running {step_name} step...\")\n        try:\n            # catch warnings to ensure deprecation warnings are printed\n            with warnings.catch_warnings(record=True) as w:\n                warnings.simplefilter(\n                    \"always\",\n                    category=PrefectDeprecationWarning,\n                )\n                warnings.simplefilter(\n                    \"always\",\n                    category=DeprecationWarning,\n                )\n                step_output = await run_step(step, upstream_outputs)\n            if w:\n                printed_messages = []\n                for warning in w:\n                    message = str(warning.message)\n                    # prevent duplicate warnings from being printed\n                    if message not in printed_messages:\n                        try:\n                            # try using rich styling\n                            print_function(message, style=\"yellow\")\n                        except Exception:\n                            # default to printing without styling\n                            print_function(message)\n                        printed_messages.append(message)\n\n            if not isinstance(step_output, dict):\n                if PREFECT_DEBUG_MODE:\n                    get_logger().warning(\n                        \"Step function %s returned unexpected type: %s\",\n                        fqn,\n                        type(step_output),\n                    )\n                continue\n            # store step output under step id to prevent clobbering\n            if inputs.get(\"id\"):\n                upstream_outputs[inputs.get(\"id\")] = step_output\n            upstream_outputs.update(step_output)\n        except Exception as exc:\n            raise StepExecutionError(f\"Encountered error while running {fqn}\") from exc\n    return upstream_outputs\n\n\ndef _get_step_fully_qualified_name_and_inputs(step: Dict) -> Tuple[str, Dict]:\n    step = deepcopy(step)\n    return step.popitem()\n", "path": "src/prefect/deployments/steps/core.py"}]}