diff --git a/pyproject.toml b/pyproject.toml index aaf5f63..9c6a3bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,6 +97,7 @@ disable = [ # Disabled on purpose: "line-too-long", # covered by black "protected-access", # covered by ruff SLF001 + "redefined-builtin", # covered by ruff "too-many-branches", # covered by ruff C901 "wrong-import-order", # covered by ruff # TODO(ssbarnea): remove temporary skips adding during initial adoption: @@ -195,6 +196,7 @@ exclude = [ ".tox", "build", "venv", - "src/subprocess_tee/_version.py" + "src/subprocess_tee/_version.py", + "src/subprocess_tee/_types.py" ] paths = ["src", "test"] diff --git a/src/subprocess_tee/__init__.py b/src/subprocess_tee/__init__.py index ffec20c..534c8f7 100644 --- a/src/subprocess_tee/__init__.py +++ b/src/subprocess_tee/__init__.py @@ -1,5 +1,7 @@ """tee like run implementation.""" +# cspell: ignore popenargs + from __future__ import annotations import asyncio @@ -21,7 +23,9 @@ __all__ = ["CompletedProcess", "__version__", "run"] if TYPE_CHECKING: - CompletedProcess = subprocess.CompletedProcess[Any] # pylint: disable=E1136 + from subprocess_tee._types import SequenceNotStr + + CompletedProcess = subprocess.CompletedProcess[Any] from collections.abc import Callable else: CompletedProcess = subprocess.CompletedProcess @@ -39,7 +43,7 @@ async def _read_stream(stream: StreamReader, callback: Callable[..., Any]) -> No async def _stream_subprocess( # noqa: C901 - args: str | list[str], + args: str | tuple[str, ...], **kwargs: Any, ) -> CompletedProcess: platform_settings: dict[str, Any] = {} @@ -136,7 +140,16 @@ def tee_func(line: bytes, sink: list[str], pipe: Any | None) -> None: ) -def run(args: str | list[str], **kwargs: Any) -> CompletedProcess: +# signature is based on stdlib +# subprocess.run() +def run( + *popenargs: str | SequenceNotStr[str], + input: bytes | str | None = None, # noqa: A002 + capture_output: bool = False, + timeout: int | None = None, + check: bool = False, + **kwargs: Any, +) -> CompletedProcess: """Drop-in replacement for subprocess.run that behaves like tee. Extra arguments added by our version: @@ -148,12 +161,31 @@ def run(args: str | list[str], **kwargs: Any) -> CompletedProcess: Raises: CalledProcessError: ... + TypeError: ... """ # run was called with a list instead of a single item but asyncio # create_subprocess_shell requires command as a single string, so # we need to convert it to string - cmd = args if isinstance(args, str) else join(args) + # breakpoint() + # if len(popenargs) == 1 and isinstance(popenargs[0], list): + # cmd = join(popenargs[0]) + # else: + args: str | tuple[str, ...] + if len(popenargs) == 0: + args = () + else: + if not isinstance(popenargs, str): # make mypy/pyright happy + raise TypeError(popenargs) + _ = popenargs[0] + if not isinstance(_, str): # make mypy/pyright happy + raise TypeError(_) + args = _ + cmd = popenargs if isinstance(popenargs, str) else join(args) + kwargs["check"] = check + kwargs["input"] = input + kwargs["timeout"] = timeout + kwargs["capture_output"] = capture_output check = kwargs.get("check", False) diff --git a/src/subprocess_tee/_types.py b/src/subprocess_tee/_types.py new file mode 100644 index 0000000..74a1231 --- /dev/null +++ b/src/subprocess_tee/_types.py @@ -0,0 +1,25 @@ +"""Internally used types.""" + +# Source from https://github.com/python/typing/issues/256#issuecomment-1442633430 +from collections.abc import Iterator, Sequence +from typing import Any, Protocol, SupportsIndex, TypeVar, overload + +_T_co = TypeVar("_T_co", covariant=True) + + +class SequenceNotStr(Protocol[_T_co]): + """Lists of strings which are not strings themselves.""" + + @overload + def __getitem__(self, index: SupportsIndex, /) -> _T_co: ... + @overload + def __getitem__(self, index: slice, /) -> Sequence[_T_co]: ... + def __contains__(self, value: object, /) -> bool: ... + def __len__(self) -> int: ... + def __iter__(self) -> Iterator[_T_co]: ... + def index( # pylint: disable=C0116 + self, value: Any, start: int = 0, stop: int = ..., / + ) -> int: ... + def count(self, value: Any, /) -> int: ... # pylint: disable=C0116 + + def __reversed__(self) -> Iterator[_T_co]: ...