8000 Controllable `subprocess` wrapper (#336) · vcs-python/libvcs@937b3ae · GitHub
[go: up one dir, main page]
Skip to content

Commit 937b3ae

Browse files
tonytony
authored
Controllable subprocess wrapper (#336)
2 parents 0c0b37e + ef26160 commit 937b3ae

File tree

5 files changed

+427
-0
lines changed

5 files changed

+427
-0
lines changed

‎CHANGES

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,12 @@ $ pip install --user --upgrade --pre libvcs
1313

1414
- _Add your latest changes from PRs here_
1515

16+
### Development
17+
18+
- {issue}`#336`: {class}`~libvcs.utils.subprocess.SubprocessCommand`: Encapsulated {mod}`subprocess`
19+
call in a {func}`dataclasses.dataclass` for introspecting, modifying, mocking and controlling
20+
execution.
21+
1622
## libvcs 0.12.0, "Nimbus" (2022-04-24)
1723

1824
### Breaking

‎docs/contributing/internals/index.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,4 +13,5 @@ stability policy.
1313
exc
1414
types
1515
query_list
16+
subprocess
1617
```

‎docs/contributing/internals/subprocess.md

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
# `libvcs.utils.subprocess`
2+
3+
```{eval-rst}
4+
.. autoapimodule:: libvcs.utils.subprocess
5+
:members:
6+
:exclude-members:
7+
StrOrBytesPath, F, args, bufsize,
8+
executable, stdin, stdout, stderr, preexec_fn, cwd,
9+
close_fds, shell, cmd, env, text, universal_newlines,
10+
startupinfo, creationflags, restore_signals, start_new_session,
11+
group, extra_groups, user, umask, pass_fds, encoding, errors,
12+
13+
```

‎libvcs/utils/subprocess.py

Lines changed: 278 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,278 @@
1+
import dataclasses
2+
import subprocess
3+
import sys
4+
from typing import (
5+
IO,
6+
Any,
7+
Callable,
8+
Literal,
9+
Mapping,
10+
Optional,
11+
Sequence,
12+
TypeVar,
13+
Union,
14+
overload,
15+
)
16+
17+
from typing_extensions import TypeAlias
18+
19+
from ..types import StrOrBytesPath
20+
21+
F = TypeVar("F", bound=Callable[..., Any])
22+
23+
24+
if sys.platform == "win32":
25+
_ENV: TypeAlias = Mapping[str, str]
26+
else:
27+
_ENV: TypeAlias = Union[
28+
Mapping[bytes, StrOrBytesPath], Mapping[str, StrOrBytesPath]
29+
]
30+
_FILE: TypeAlias = Union[None, int, IO[Any]]
31+
_TXT: TypeAlias = Union[bytes, str]
32+
#: Command
33+
_CMD: TypeAlias = Union[StrOrBytesPath, Sequence[StrOrBytesPath]]
34+
35+
36+
@dataclasses.dataclass
37+
class SubprocessCommand:
38+
"""Encapsulate a :mod:`subprocess` request. Inspect, mutate, control before invocation.
39+
40+
Attributes
41+
----------
42+
args : _CMD
43+
A string, or a sequence of program arguments.
44+
45+
bufsize : int
46+
supplied as the buffering argument to the open() function when creating the
47+
stdin/stdout/stderr pipe file objects
48+
49+
executable : Optional[StrOrBytesPath]
50+
A replacement program to execute.
51+
52+
stdin : _FILE
53+
standard output for executed program
54+
55+
stdout :
56+
standard output for executed program
57+
58+
stderr :
59+
standard output for executed program
60+
61+
close_fds : Controls closing or inheriting of file descriptors.
62+
63+
shell : If true, the command will be executed through the shell.
64+
65+
cwd : Sets the current directory before the child is executed.
66+
67+
env : Defines the environment variables for the new process.
68+
69+
text :
70+
If ``True``, decode stdin, stdout and stderr using the given encoding (if set)
71+
or the system default otherwise.
72+
73+
universal_newlines :
74+
Alias of text, provided for backwards compatibility.
75+
76+
startupinfo :
77+
Windows only
78+
79+
creationflags :
80+
Windows only
81+
82+
preexec_fn :
83+
(POSIX only) An object to be called in the child process just before the child
84+
is executed.
85+
86+
restore_signals :
87+
POSIX only
88+
89+
start_new_session :
90+
POSIX only
91+
92+
group :
93+
POSIX only
94+
95+
extra_groups :
96+
POSIX only
97+
98+
user :
99+
POSIX only
100+
101+
umask :
102+
POSIX only
103+
104+
pass_fds :
105+
POSIX only
106+
107+
encoding :
108+
Text mode encoding to use for file objects stdin, stdout and stderr.
109+
110+
errors :
111+
Text mode error handling to use for file objects stdin, stdout and stderr.
112+
113+
Examples
114+
--------
115+
>>> cmd = SubprocessCommand("ls")
116+
>>> cmd.args
117+
'ls'
118+
119+
With ``shell=True``:
120+
121+
>>> cmd = SubprocessCommand("ls -l", shell=True)
122+
>>> cmd.shell
123+
True
124+
>>> cmd.args
125+
'ls -l'
126+
>>> cmd.check_call()
127+
0
128+
"""
129+
130+
args: _CMD
131+
bufsize: int = -1
132+
executable: Optional[StrOrBytesPath] = None
133+
stdin: _FILE = None
134+
stdout: _FILE = None
135+
stderr: _FILE = None
136+
preexec_fn: Optional[Callable[[], Any]] = None
137+
close_fds: bool = True
138+
shell: bool = False
139+
cwd: Optional[StrOrBytesPath] = None
140+
env: Optional[_ENV] = None
141+
142+
# Windows
143+
creationflags: int = 0
144+
startupinfo: Optional[Any] = None
145+
146+
# POSIX-only
147+
restore_signals: bool = True
148+
start_new_session: bool = False
149+
pass_fds: Any = ()
150+
umask: int = -1
151+
if sys.version_info >= (3, 10):
152+
pipesize: int = -1
153+
user: Optional[str] = None
154+
group: Optional[str] = None
155+
extra_groups: Optional[list[str]] = None
156+
157+
# Alias of text, for backwards compatibility
158+
universal_newlines: Optional[bool] = None
159+
text: Optional[Literal[True]] = None
160+
161+
# Text mode encoding and error handling to use for file objects
162+
# stdin, stdout, stderr
163+
encoding: Optional[str] = None
164+
errors: Optional[str] = None
165+
166+
def Popen(self, **kwargs) -> subprocess.Popen:
167+
"""Run commands :class:`subprocess.Popen`, optionally overrides via kwargs.
168+
169+
Parameters
170+
----------
171+
**kwargs : dict, optional
172+
Overrides existing attributes for :class:`subprocess.Popen`
173+
174+
Examples
175+
--------
176+
>>> cmd = SubprocessCommand(args=['echo', 'hello'])
177+
>>> proc = cmd.Popen(stdout=subprocess.PIPE)
178+
>>> proc.communicate() # doctest: +SKIP
179+
"""
180+
return subprocess.Popen(**dataclasses.replace(self, **kwargs).__dict__)
181+
182+
def check_call(self, **kwargs) -> int:
183+
"""Run command :func:`subprocess.check_call`, optionally overrides via kwargs.
184+
185+
Parameters
186+
----------
187+
**kwargs : dict, optional
188+
Overrides existing attributes for :func:`subprocess.check_call`
189+
190+
Examples
191+
--------
192+
>>> cmd = SubprocessCommand(args=['echo', 'hello'])
193+
>>> cmd.check_call(stdout=subprocess.PIPE)
194+
0
195+
"""
196+
return subprocess.check_call(**dataclasses.replace(self, **kwargs).__dict__)
197+
198+
@overload
199+
def check_output(self, input: Optional[str] = None, **kwargs) -> str:
200+
...
201+
202+
@overload
203+
def check_output(self, input: Optional[bytes] = None, **kwargs) -> bytes:
204+
...
205+
206+
def check_output(
207+
self, input: Optional[Union[str, bytes]] = None, **kwargs
208+
) -> Union[bytes, str]:
209+
r"""Run command :func:`subprocess.check_output`, optionally overrides via kwargs.
210+
211+
Parameters
212+
----------
213+
input : Union[bytes, str], optional
214+
pass string to subprocess's stdin. Bytes by default, str in text mode.
215+
216+
Text mode is triggered by setting any of text, encoding, errors or
217+
universal_newlines.
218+
**kwargs : dict, optional
219+
Overrides existing attributes for :func:`subprocess.check_output`
220+
221+
Examples
222+
--------
223+
>>> cmd = SubprocessCommand(args=['echo', 'hello'])
224+
>>> proc = cmd.check_output(shell=True)
225+
226+
From :mod:`subprocess`:
227+
228+
>>> import subprocess
229+
>>> cmd = SubprocessCommand(
230+
... ["/bin/sh", "-c", "ls -l non_existent_file ; exit 0"])
231+
>>> cmd.check_output(stderr=subprocess.STDOUT)
232+
b"ls: ...non_existent_file...: No such file or directory\n"
233+
234+
>>> cmd = SubprocessCommand(["sed", "-e", "s/foo/bar/"])
235+
>>> cmd.check_output(input=b"when in the course of fooman events\n")
236+
b'when in the course of barman events\n'
237+
"""
238+
params = dataclasses.replace(self, **kwargs).__dict__
239+
params.pop("stdout")
240+
return subprocess.check_output(input=input, **params)
241+
242+
def run(
243+
self,
244+
input: Optional[Union[str, bytes]] = None,
245+
timeout: Optional[int] = None,
246+
check: bool = False,
247+
capture_output: bool = False,
248+
**kwargs,
249+
) -> subprocess.CompletedProcess:
250+
"""Run command in :func:`subprocess.run`, optionally overrides via kwargs.
251+
252+
Parameters
253+
----------
254+
input : Union[bytes, str], optional
255+
pass string to subprocess's stdin. Bytes by default, str in text mode.
256+
257+
Text mode is triggered by setting any of text, encoding, errors or
258+
universal_newlines.
259+
260+
check : bool
261+
If True and the exit code was non-zero, it raises a
262+
:exc:`subprocess.CalledProcessError`. The CalledProcessError object will
263+
have the return code in the returncode attribute, and output & stderr
264+
attributes if those streams were captured.
265+
266+
timeout : int
267+
If given, and the process takes too long, a :exc:`subprocess.TimeoutExpired`
268+
269+
**kwargs : dict, optional
270+
Overrides existing attributes for :func:`subprocess.run`
271+
"""
272+
return subprocess.run(
273+
**dataclasses.replace(self, **kwargs).__dict__,
274+
input=input,
275+
capture_output=capture_output,
276+
timeout=timeout,
277+
check=check,
278+
)

0 commit comments

Comments
 (0)
0