Repository: PrefectHQ/prefect
Pull request: pass through kwargs for `run_shell_process`
closes https://github.com/PrefectHQ/prefect/issues/15816

Problem statement:
run_shell_process should support passing the current working directory or environment
### Describe the current behavior

The `run_shell_process` flow currently assumes it is running from within the current working directory and with the default environment variables. When the `run_shell_process` flow is either run outside the target working directory, or as a child flow, those values cannot be passed as arguments.

### Describe the proposed behavior

The `run_shell_process` flow should accept additional optional arguments for `cwd` and `env` that are passed to the subprocess.Popen call.

```python
@flow
def run_shell_process(
    command: str,
    log_output: bool = True,
    stream_stdout: bool = False,
    log_stderr: bool = False,
    cwd: str = None,
    env: Dict[str, str] = None,
):
```

### Example Use

```python
run_shell_process(
    command="touch test.txt",
    cwd="/tmp/test",
    env=os.environ.copy(),
    stream_stdout=True,
    return_state=True,
)
```

### Additional context

`run_shell_process` is being run as a child flow within a Prefect (v3) flow.

Hints from issue discussion:
hi @Jared314 - thanks for the issue! the PR linked above should allow this - let me know if you have any concerns!

Commit message: slightly cleaner

Patch to review:
diff --git a/src/prefect/cli/shell.py b/src/prefect/cli/shell.py
index 6ca13f50bf6c..83a767402cfc 100644
--- a/src/prefect/cli/shell.py
+++ b/src/prefect/cli/shell.py
@@ -8,7 +8,7 @@
 import subprocess
 import sys
 import threading
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 
 import typer
 from typing_extensions import Annotated
@@ -62,6 +62,7 @@ def run_shell_process(
     log_output: bool = True,
     stream_stdout: bool = False,
     log_stderr: bool = False,
+    popen_kwargs: Optional[Dict[str, Any]] = None,
 ):
     """
     Asynchronously executes the specified shell command and logs its output.
@@ -70,12 +71,11 @@ def run_shell_process(
     It handles both the execution of the command and the collection of its output for logging purposes.
 
     Args:
-        command (str): The shell command to execute.
-        log_output (bool, optional): If True, the output of the command (both stdout and stderr) is logged to Prefect.
-                                     Defaults to True
-        stream_stdout (bool, optional): If True, the stdout of the command is streamed to Prefect logs. Defaults to False.
-        log_stderr (bool, optional): If True, the stderr of the command is logged to Prefect logs. Defaults to False.
-
+        command: The shell command to execute.
+        log_output: If True, the output of the command (both stdout and stderr) is logged to Prefect.
+        stream_stdout: If True, the stdout of the command is streamed to Prefect logs.
+        log_stderr: If True, the stderr of the command is logged to Prefect logs.
+        popen_kwargs: Additional keyword arguments to pass to the `subprocess.Popen` call.
 
     """
 
@@ -91,6 +91,7 @@ def run_shell_process(
         text=True,
         bufsize=1,
         universal_newlines=True,
+        **(popen_kwargs or {}),
     ) as proc:
         # Create threads for collecting stdout and stderr
         if stream_stdout:
diff --git a/tests/cli/test_shell.py b/tests/cli/test_shell.py
index c85db78a03b8..0a209ecb4cfc 100644
--- a/tests/cli/test_shell.py
+++ b/tests/cli/test_shell.py
@@ -1,5 +1,7 @@
+import os
 from unittest.mock import AsyncMock, patch
 
+from prefect.cli.shell import run_shell_process
 from prefect.testing.cli import invoke_and_assert
 from prefect.utilities.asyncutils import run_sync_in_worker_thread
 
@@ -145,3 +147,33 @@ async def test_shell_runner_integration(monkeypatch):
         )
 
         runner_mock.assert_awaited_once_with(run_once=True)
+
+
+class TestRunShellProcess:
+    def test_run_shell_process_basic(self, tmp_path):
+        """Test basic command execution"""
+        test_file = tmp_path / "test.txt"
+        run_shell_process(f"touch {test_file}")
+        assert test_file.exists()
+
+    def test_run_shell_process_with_cwd(self, tmp_path):
+        """Test command execution with custom working directory"""
+        subdir = tmp_path / "subdir"
+        subdir.mkdir()
+        test_file = "test.txt"
+
+        run_shell_process(f"touch {test_file}", popen_kwargs={"cwd": str(subdir)})
+
+        assert (subdir / test_file).exists()
+
+    def test_run_shell_process_with_env(self, tmp_path):
+        """Test command execution with custom environment variables"""
+        custom_env = os.environ.copy()
+        custom_env["TEST_VAR"] = "hello"
+
+        run_shell_process(
+            "echo $TEST_VAR > output.txt",
+            popen_kwargs={"env": custom_env, "cwd": str(tmp_path)},
+        )
+
+        assert (tmp_path / "output.txt").read_text().strip() == "hello"


Code review comments:
diff_hunk: @@ -91,6 +91,7 @@ def run_shell_process(
         text=True,
path: src/prefect/cli/shell.py
text: @user1:
I think we should move the existing hard-coded kwargs to defaults on the `popen_kwargs` dict to avoid user-provided kwargs conflicting with the internally used ones.

@author:
ahh thats a good point - will update