refactor!(run): Match subprocess.Popen

Author: tony

libvcs/_internal/run.py

@@ -13,7 +13,20 @@
 import os
 import subprocess
 import sys
-from typing import Optional, Protocol, Union
+from typing import (
+    IO,
+    Any,
+    Callable,
+    Iterable,
+    Mapping,
+    Optional,
+    Protocol,
+    Sequence,
+    Union,
+    overload,
+)
+
+from typing_extensions import TypeAlias
 
 from .. import exc
 from ..types import StrOrBytesPath
@@ -143,15 +156,97 @@ def __call__(self, output: Union[str, bytes], timestamp: datetime.datetime):
         ...
 
 
+if sys.platform == "win32":
+    _ENV: TypeAlias = Mapping[str, str]
+else:
+    _ENV: TypeAlias = Mapping[bytes, StrOrBytesPath] | Mapping[str, StrOrBytesPath]
+
+_CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]]
+_FILE: TypeAlias = None | int | IO[Any]
+
+# def run(
+#     args: _CMD,
+#     bufsize: int = -1,
+#     executable: StrOrBytesPath | None = None,
+#     stdin: _FILE | None = None,
+#     stdout: _FILE | None = None,
+#     stderr: _FILE | None = None,
+#     preexec_fn: Callable[[], Any] | None = None,
+#     close_fds: bool = True,
+#     shell: bool = False,
+#     cwd: StrOrBytesPath | None = None,
+#     env: _ENV | None = None,
+#     universal_newlines: bool = False,
+#     startupinfo: Any | None = None,
+#     creationflags: int = 0,
+#     restore_signals: bool = True,
+#     start_new_session: bool = False,
+#     pass_fds: Any = None,
+#     *,
+#     text: bool | None = None,
+#     encoding: str = "utf-8",
+#     errors: str | None = None,
+#     user: str | int | None = None,
+#     group: str | int | None = None,
+#     extra_groups: Iterable[str | int] | None = None,
+#     umask: int = -1,
+#     pipesize: int = -1,
+#
+#
+#     # custom
+#     log_in_real_time: bool = True,
+#     check_returncode: bool = True,
+#     callback: Optional[ProgressCallbackProtocol] = None,
+# ):
+
+
+@overload
 def run(
-    cmd: Union[str, list[str]],
+    args: _CMD,
+    bufsize: int = ...,
+    executable: StrOrBytesPath | None = ...,
+    stdin: _FILE | None = ...,
+    stdout: _FILE | None = ...,
+    stderr: _FILE | None = ...,
+    preexec_fn: Callable[[], Any] | None = ...,
+    close_fds: bool = ...,
+    shell: bool = ...,
+    cwd: Optional[StrOrBytesPath] = ...,
+    env: _ENV | None = ...,
+    universal_newlines: bool = ...,
+    startupinfo: Any | None = ...,
+    creationflags: int = ...,
+    restore_signals: bool = ...,
+    start_new_session: bool = ...,
+    pass_fds: Any = ...,
+    *,
+    text: bool | None = ...,
+    encoding: str = "utf-8",
+    errors: str | None = ...,
+    user: str | int | None = ...,
+    group: str | int | None = ...,
+    extra_groups: Iterable[str | int] | None = ...,
+    umask: int = ...,
+    pipesize: int = ...,
+    # custom
+    log_in_real_time: bool = True,
+    check_returncode: bool = True,
+    callback: Optional[ProgressCallbackProtocol] = None,
+):
+    ...
+
+
+def run(
+    # args: Union[str, list[str]],
+    args: _CMD,
     shell: bool = False,
     cwd: Optional[StrOrBytesPath] = None,
+    *,
+    # custom
     log_in_real_time: bool = True,
     check_returncode: bool = True,
     callback: Optional[ProgressCallbackProtocol] = None,
-    *args,
-    **kwargs
+    **kwargs,
 ):
     """Run 'cmd' in a shell and return the combined contents of stdout and
     stderr (Blocking).  Throws an exception if the command exits non-zero.
@@ -189,13 +284,12 @@ def progress_cb(output, timestamp):
             run(['git', 'pull'], callback=progress_cb)
     """
     proc = subprocess.Popen(
-        cmd,
+        args,
         shell=shell,
         stderr=subprocess.PIPE,
         stdout=subprocess.PIPE,
         cwd=cwd,
-        *args,
-        **kwargs
+        **kwargs,
     )
 
     all_output = []
@@ -220,5 +314,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,