diff --git a/CHANGES b/CHANGES index 1ff5be8b7..405da4959 100644 --- a/CHANGES +++ b/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 diff --git a/libvcs/_internal/run.py b/libvcs/_internal/run.py index 7cadbed23..b83280ba8 100644 --- a/libvcs/_internal/run.py +++ b/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 diff --git a/libvcs/cmd/git.py b/libvcs/cmd/git.py index 47ef047c3..cc7491ce3 100644 --- a/libvcs/cmd/git.py +++ b/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, diff --git a/libvcs/cmd/hg.py b/libvcs/cmd/hg.py index 0b116010a..aed789df0 100644 --- a/libvcs/cmd/hg.py +++ b/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, diff --git a/libvcs/cmd/svn.py b/libvcs/cmd/svn.py index 836f8ded8..9bfbd4dcc 100644 --- a/libvcs/cmd/svn.py +++ b/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,