chore(run): Pass through to subprocess.Popen (#361)

Author: tony

CHANGES

@@ -17,6 +17,10 @@ $ pip install --user --upgrade --pre libvcs
 
 - {issue}`343`: `libvcs.cmd.core` (including {func}`~libvcs._internal.run.run`) have been moved to
   `libvcs._internal.run`. It will be supported as an unstable, internal API.
+- {issue}`361`: {class}`~libvcs._internal.run.run`'s params are now a pass-through to
+  {class}`subprocess.Popen`.
+
+  - `run(cmd, ...)` is now `run(args, ...)` to match `Popen`'s convention.
 
 ### What's new
 

libvcs/_internal/run.py

@@ -13,7 +13,19 @@
 import os
 import subprocess
 import sys
-from typing import Optional, Protocol, Union
+from typing import (
+    IO,
+    Any,
+    Callable,
+    Iterable,
+    Mapping,
+    Optional,
+    Protocol,
+    Sequence,
+    Union,
+)
+
+from typing_extensions import TypeAlias
 
 from .. import exc
 from ..types import StrOrBytesPath
@@ -143,28 +155,65 @@ def __call__(self, output: Union[str, bytes], timestamp: datetime.datetime):
         ...
 
 
+if sys.platform == "win32":
+    _ENV: TypeAlias = Mapping[str, str]
+else:
+    _ENV: TypeAlias = Union[
+        Mapping[bytes, StrOrBytesPath], Mapping[str, StrOrBytesPath]
+    ]
+
+_CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]]
+_FILE: TypeAlias = Optional[Union[int, IO[Any]]]
+
+
 def run(
-    cmd: Union[str, list[str]],
+    args: _CMD,
+    bufsize: int = -1,
+    executable: Optional[StrOrBytesPath] = None,
+    stdin: Optional[_FILE] = None,
+    stdout: Optional[_FILE] = None,
+    stderr: Optional[_FILE] = None,
+    preexec_fn: Optional[Callable[[], Any]] = None,
+    close_fds: bool = True,
     shell: bool = False,
     cwd: Optional[StrOrBytesPath] = None,
+    env: Optional[_ENV] = None,
+    universal_newlines: Optional[bool] = None,
+    startupinfo: Optional[Any] = None,
+    creationflags: int = 0,
+    restore_signals: bool = True,
+    start_new_session: bool = False,
+    pass_fds: Any = (),
+    *,
+    text: Optional[bool] = None,
+    encoding: Optional[str] = None,
+    errors: Optional[str] = None,
+    user: Optional[Union[str, int]] = None,
+    group: Optional[Union[str, int]] = None,
+    extra_groups: Optional[Iterable[Union[str, int]]] = None,
+    umask: int = -1,
+    # Not until sys.version_info >= (3, 10)
+    # pipesize: int = -1,
+    # custom
     log_in_real_time: bool = True,
     check_returncode: bool = True,
     callback: Optional[ProgressCallbackProtocol] = None,
 ):
-    """Run 'cmd' in a shell and return the combined contents of stdout and
-    stderr (Blocking).  Throws an exception if the command exits non-zero.
+    """Run 'args' in a shell and return the combined contents of stdout and
+    stderr (Blocking). Throws an exception if the command exits non-zero.
+
+    Keyword arguments are passthrough to {class}`subprocess.Popen`.
 
     Parameters
     ----------
-    cmd : list or str, or single str, if shell=True
+    args : list or str, or single str, if shell=True
        the command to run
 
     shell : boolean
         boolean indicating whether we are using advanced shell
         features. Use only when absolutely necessary, since this allows a lot
         more freedom which could be exploited by malicious code. See the
-        warning here:
-        http://docs.python.org/library/subprocess.html#popen-constructor
+        warning here: http://docs.python.org/library/subprocess.html#popen-constructor
 
     cwd : str
         dir command is run from. Defaults to ``path``.
@@ -187,11 +236,30 @@ def progress_cb(output, timestamp):
             run(['git', 'pull'], callback=progress_cb)
     """
     proc = subprocess.Popen(
-        cmd,
+        args,
+        bufsize=bufsize,
+        executable=executable,
+        stdin=stdin,
+        stdout=stdout or subprocess.PIPE,
+        stderr=stderr or subprocess.PIPE,
+        preexec_fn=preexec_fn,
+        close_fds=close_fds,
         shell=shell,
-        stderr=subprocess.PIPE,
-        stdout=subprocess.PIPE,
         cwd=cwd,
+        env=env,
+        universal_newlines=universal_newlines,
+        startupinfo=startupinfo,
+        creationflags=creationflags,
+        restore_signals=restore_signals,
+        start_new_session=start_new_session,
+        pass_fds=pass_fds,
+        text=text,
+        encoding=encoding,
+        errors=errors,
+        user=user,
+        group=group,
+        extra_groups=extra_groups,
+        umask=umask,
     )
 
     all_output = []
@@ -216,5 +284,5 @@ def progress_cb(output, timestamp):
         all_output = console_to_str(b"".join(stderr_lines))
     output = "".join(all_output)
     if code != 0 and check_returncode:
-        raise exc.CommandError(output=output, returncode=code, cmd=cmd)
+        raise exc.CommandError(output=output, returncode=code, cmd=args)
     return output

libvcs/cmd/git.py

@@ -181,7 +181,7 @@ def run(
         if no_optional_locks is True:
             cli_args.append("--no-optional-locks")
 
-        return run(cmd=cli_args, **kwargs)
+        return run(args=cli_args, **kwargs)
 
     def clone(
         self,

libvcs/cmd/hg.py

@@ -2,9 +2,10 @@
 import pathlib
 from typing import Optional, Sequence, Union
 
-from ..types import StrOrBytesPath, StrPath
 from libvcs._internal.run import run
 
+from ..types import StrOrBytesPath, StrPath
+
 _CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]]
 
 
@@ -156,7 +157,7 @@ def run(
         if help is True:
             cli_args.append("--help")
 
-        return run(cmd=cli_args, **kwargs)
+        return run(args=cli_args, **kwargs)
 
     def clone(
         self,

libvcs/cmd/svn.py

@@ -1,9 +1,10 @@
 import pathlib
 from typing import Literal, Optional, Sequence, Union
 
-from ..types import StrOrBytesPath, StrPath
 from libvcs._internal.run import run
 
+from ..types import StrOrBytesPath, StrPath
+
 _CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]]
 
 DepthLiteral = Union[Literal["infinity", "empty", "files", "immediates"], None]
@@ -107,7 +108,7 @@ def run(
         if config_option is not None:
             cli_args.append("--config-option {config_option}")
 
-        return run(cmd=cli_args, **kwargs)
+        return run(args=cli_args, **kwargs)
 
     def checkout(
         self,