shellous
Async Processes and Pipelines
shellous provides a concise API for running subprocesses using asyncio. It is similar to and inspired by sh.
import asyncio
from shellous import sh
async def main():
result = await sh("echo", "hello")
print(result)
asyncio.run(main())
Benefits
- Run programs asynchronously in a single line.
- Redirect stdin, stdout and stderr to files, memory buffers, async streams or loggers.
- Iterate asynchronously over subprocess output.
- Set timeouts and reliably cancel running processes.
- Run a program with a pseudo-terminal (pty).
- Use send() and expect() to manually control a subprocess.
- Construct pipelines and use process substitution directly from Python (no shell required).
- Runs on Linux, MacOS, FreeBSD and Windows.
- Monitor processes being started and stopped with
audit_callbackAPI.
Requirements
- Requires Python 3.10 or later.
- Requires an asyncio event loop.
- Pseudo-terminals require a Unix system.
- Process substitution requires a Unix system with /dev/fd support.
Running a Command
The tutorial in this README uses the asyncio REPL built into Python. In these examples, >>>
is the REPL prompt.
Start the asyncio REPL by typing python3 -m asyncio, and import sh from the shellous module:
>>> from shellous import sh
Here's a command that runs echo "hello, world".
>>> await sh("echo", "hello, world")
'hello, world\n'
The first argument to sh is the program name. It is followed by zero or more arguments. Each argument will be
converted to a string. If an argument is a list or tuple, it is flattened recursively.
>>> await sh("echo", 1, 2, [3, 4, (5, 6)])
'1 2 3 4 5 6\n'
A command does not run until you await it. When you run a command using await, it returns the value of the standard output interpreted as a UTF-8 string.
It is safe to await the same command object more than once.1
Here, we create our own echo command with "-n" to omit the newline. Note, echo("abc") will run the same command as echo -n "abc".
>>> echo = sh("echo", "-n")
>>> await echo("abc")
'abc'
Commands are immutable objects that represent a program invocation: program name, arguments, environment
variables, redirection operators and other settings. When you use a method to modify a Command, you are
returning a new Command object. The original object is unchanged.1
You can wrap your commands in a function to improve type safety:
>>> from shellous import Command
>>> def exclaim(word: str) -> Command[str]:
... return sh("echo", "-n", f"{word}!!")
...
>>> await exclaim("Oh")
'Oh!!'
The type hint Command[str] indicates that the command returns a str.
generator objects cannot be reused. Shellous will detect this case and raise an error.
Arguments
Commands use positional arguments only; keyword arguments are not supported.
In most cases, shellous automatically converts Python objects passed as command arguments to str or bytes. As described above, the list and tuple types are an exception; they are recursively flattened before their elements are converted to strings.
Dicts, sets, and generator types are not supported as arguments. Their string format doesn't make sense as a command line argument.
Results
When a command completes successfully, it returns the standard output (or "" if stdout is redirected). For a more detailed response, you can specify that the command should return a Result object by using the .result modifier:
>>> await echo.result("abc")
Result(exit_code=0, output_bytes=b'abc', error_bytes=b'', cancelled=False, encoding='utf-8')
A Result object contains the command's exit_code in addition to its output. A Result is True if
the command's exit_code is zero. You can access the string value of the output using the .output property:
if result := await sh.result("cat", "some-file"):
output = result.output
else:
print(f"Command failed with exit_code={result.exit_code})
You can retrieve the string value of the standard error using the .error property. (By default, only the
first 1024 bytes of standard error is stored.)
If a command was terminated by a signal, the exit_code will be the negative signal number.
The return value of sh.result("cmd", ...) uses the type hint Command[Result].
ResultError
If you are not using the .result modifier and a command fails, it raises a ResultError exception:
>>> await sh("cat", "does_not_exist")
Traceback (most recent call last):
...
shellous.result.ResultError: Result(exit_code=1, output_bytes=b'', error_bytes=b'cat: does_not_exist: No such file or directory\n', cancelled=False, encoding='utf-8')
The ResultError exception contains a Result object with the exit_code and the first 1024 bytes of standard error.
In some cases, you want to ignore certain exit code values. That is, you want to treat them as if they are
normal. To do this, you can set the exit_codes option:
>>> await sh("cat", "does_not_exist").set(exit_codes={0,1})
''
If there is a problem launching a process, shellous can also raise a separate FileNotFoundError or PermissionError exception.
Async For
Using await to run a command collects the entire output of the command in memory before returning it. You
can also iterate over the output lines as they arrive using async for.
>>> [line async for line in echo("hi\n", "there")]
['hi\n', ' there']
Use an async for loop when you want to examine the stream of output from a command, line by line. For example, suppose you want to run tail on a log file.
async for line in sh("tail", "-f", "/var/log/syslog"):
if "ERROR" in line:
print(line.rstrip())
Async With
You can use a command as an asynchronous context manager. There are two ways to run a program using a context manager: a low-level API and a high-level API.
Byte-by-Byte (Low Level)
Use async with directly when you need byte-by-byte
control over the individual streams: stdin, stdout and stderr. To control a standard stream, you
must tell shellous to "capture" it (For more on this, see Redirection.)
cmd = sh("cat").stdin(sh.CAPTURE).stdout(sh.CAPTURE)
async with cmd as run:
run.stdin.write(b"abc")
run.stdin.close()
print(await run.stdout.readline())
result = run.result()
The streams run.stdout and run.stderr are asyncio.StreamReader objects. The stream run.stdin
is an asyncio.StreamWriter object. If we didn't specify that stdin/stdout are sh.CAPTURE, the
streams run.stdin and run.stdout would be
None.
The return value of run.result() is a Result object. Depending on the command settings, this
function may raise a ResultError on a non-zero exit code.
:warning: When reading or writing individual streams, you are responsible for managing reads and writes so they don't deadlock. You may use
run.create_taskto schedule a concurrent task.
You can also use async with to run a server. When you do so, you must tell the server
to stop using run.cancel(). Otherwise, the context manager will wait forever for the process to exit.
async with sh("some-server") as run:
# Send commands to the server here...
# Manually signal the server to stop.
run.cancel()
Prompt with Send/Expect (High Level API)
Use the prompt() method to control a process using send and expect. The prompt()
method returns an asynchronous context manager (the Prompt class) that facilitates reading and
writing strings and matching regular expressions.
cmd = sh("cat").set(pty=True)
async with cmd.prompt() as client:
await client.send("abc")
output, _ = await client.expect("\r\n")
print(output)
The Prompt API automatically captures stdin and stdout.
Here is another example of controlling a bash co-process running in a docker container.
async def list_packages():
"Run bash in an ubuntu docker container and list packages."
bash_prompt = re.compile("root@[0-9a-f]+:/[^#]*# ")
cmd = sh("docker", "run", "-it", "--rm", "-e", "TERM=dumb", "ubuntu")
async with cmd.set(pty=True).prompt(bash_prompt, timeout=3) as cli:
# Read up to first prompt.
await cli.expect()
# Disable echo. The `command()` method combines send *and* expect methods.
await cli.command("stty -echo")
# Return list of packages.
result = await cli.command("apt-cache pkgnames")
return result.strip().split("\r\n")
# You can check the result object's exit code. You can only
# access `cli.result` outside the `async with` block.
assert cli.result.exit_code == 0
The prompt() API does not raise a ResultError when a command exits with an error status.
Typically, you'll see an EOFError when you were expecting to read a response. You can check the
exit status by retrieving the Prompt's result property outside of the async with block.
Redirection
shellous supports the redirection operators | and >>. They work similar to how they work in
the unix shell. Shellous does not support use of < or > for redirection. Instead, replace these
with |.
To redirect to or from a file, use a pathlib.Path object. Alternatively, you can redirect input/output
to a StringIO object, an open file, a Logger, or use a special redirection constant like sh.DEVNULL.
:warning: When combining the redirect operators with
await, you must use parentheses;awaithas higher precedence than|and>>.
Redirecting Standard Input
To redirect standard input, use the pipe operator | with the argument on the left-side.
Here is an example that passes the string "abc" as standard input.
>>> cmd = "abc" | sh("wc", "-c")
>>> await cmd
' 3\n'
To read input from a file, use a Path object from pathlib.
>>> from pathlib import Path
>>> cmd = Path("LICENSE") | sh("wc", "-l")
>>> await cmd
' 201\n'
Shellous supports different STDIN behavior when using different Python types.
| Python Type | Behavior as STDIN |
|---|---|
| str | Read input from string object. |
| bytes, bytearray | Read input from bytes object. |
| Path | Read input from file specified by Path. |
| File, StringIO, ByteIO | Read input from open file object. |
| int | Read input from existing file descriptor. |
| asyncio.StreamReader | Read input from a StreamReader. |
| AsyncGenerator[str | bytes, None] | Read input from an async generator. |
| sh.DEVNULL | Read input from /dev/null. |
| sh.INHERIT | Read input from existing sys.stdin. |
| sh.CAPTURE | You will write to stdin interactively. |
Redirecting Standard Output
To redirect standard output, use the pipe operator | with the argument on the right-side. Here is an
example that writes to a temporary file.
>>> output_file = Path("/tmp/output_file")
>>> cmd = sh("echo", "abc") | output_file
>>> await cmd
''
>>> output_file.read_bytes()
b'abc\n'
To redirect standard output with append, use the >> operator.
>>> cmd = sh("echo", "def") >> output_file
>>> await cmd
''
>>> output_file.read_bytes()
b'abc\ndef\n'
Shellous supports different STDOUT behavior when using different Python types.
| Python Type | Behavior as STDOUT/STDERR |
|---|---|
| Path | Write output to the file path specified by Path. |
| bytearray | Write output to a mutable byte array. |
| File, StringIO, ByteIO | Write output to an open file object. |
| int | Write output to existing file descriptor at its current position. â—† |
| logging.Logger | Log each line of output. â—† |
| asyncio.StreamWriter | Write output to StreamWriter. â—† |
| AsyncGenerator[None, bytes] | Write output to an async generator. â—† |
| sh.CAPTURE | Capture output for async with. â—† |
| sh.DEVNULL | Write output to /dev/null. â—† |
| sh.INHERIT | Write output to existing sys.stdout or sys.stderr. â—† |
â—† For these types, there is no difference between using | and >>.
Shellous does not support redirecting standard output/error to a plain str or bytes object.
If you intend to redirect output to a file, you must use a pathlib.Path object.
Redirecting Standard Error
By default, the first 1024 bytes read from standard error are stored in the Result object.
Any further bytes are discarded. You can change the 1024 byte limit using the error_limit
option.
To redirect standard error, use the stderr method. Standard error supports the
same Python types as standard output. To append, set append=True in the stderr method.
To redirect stderr to the same place as stdout, use the sh.STDOUT constant. If you also
redirect stdout to sh.DEVNULL, you will only receive the standard error.
>>> cmd = sh("cat", "does_not_exist").stderr(sh.STDOUT)
>>> await cmd.set(exit_codes={0,1})
'cat: does_not_exist: No such file or directory\n'
To redirect standard error to the hosting program's sys.stderr, use the sh.INHERIT redirect
option.
>>> cmd = sh("cat", "does_not_exist").stderr(sh.INHERIT)
>>> await cmd
cat: does_not_exist: No such file or directory
Traceback (most recent call last):
...
shellous.result.ResultError: Result(exit_code=1, output_bytes=b'', error_bytes=b'', cancelled=False, encoding='utf-8')
If you redirect stderr, it will no longer be stored in the Result object, and the error_limit option
will not apply.
Default Redirections
For regular commands, the default redirections are:
- Standard input is read from the empty string ("").
- Standard out is buffered and stored in the Result object (BUFFER).
- First 1024 bytes of standard error is buffered and stored in the Result object (BUFFER).
However, the default redirections are adjusted when using a pseudo-terminal (pty):
- Standard input is captured and ignored (CAPTURE).
- Standard out is buffered and stored in the Result object (BUFFER).
- Standard error is redirected to standard output (STDOUT).
When you use the Prompt API, the standard input and standard output are automatically redirected to CAPTURE.
Pipelines
You can create a pipeline by combining commands using the | operator. A pipeline feeds the standard out of one process into the next process as standard input. Here is the shellous
equivalent to the bash command: ls | grep README
>>> pipe = sh("ls") | sh("grep", "README")
>>> await pipe
'README.md\n'
A pipeline returns a Result if the last command in the pipeline has the .result modifier. To set other
options like encoding for a Pipeline, set them on the last command.
>>> pipe = sh("ls") | sh("grep", "README").result
>>> await pipe
Result(exit_code=0, output_bytes=b'README.md\n', error_bytes=b'', cancelled=False, encoding='utf-8')
Error reporting for a pipeline is implemented similar to using the -o pipefail shell option.
Pipelines support the same await/async for/async with operations that work on a single command, including
the Prompt API.
>>> [line.strip() async for line in pipe]
['README.md']
Process Substitution (Unix Only)
You can pass a shell command as an argument to another. Here is the shellous equivalent to the bash
command: grep README <(ls).
>>> cmd = sh("grep", "README", sh("ls"))
>>> await cmd
'README.md\n'
Use .writable to write to a command instead.
>>> buf = bytearray()
>>> cmd = sh("ls") | sh("tee", sh("grep", "README").writable | buf) | sh.DEVNULL
>>> await cmd
''
>>> buf
bytearray(b'README.md\n')
The above example is equivalent to ls | tee >(grep README > buf) > /dev/null.
Timeouts
You can specify a timeout using the timeout option. If the timeout expires, shellous will raise
a TimeoutError.
>>> await sh("sleep", 60).set(timeout=0.1)
Traceback (most recent call last):
...
TimeoutError
Timeouts are just a special case of cancellation. When a command is cancelled, shellous terminates
the running process and raises a CancelledError.
>>> t = asyncio.create_task(sh("sleep", 60).coro())
>>> t.cancel()
True
>>> await t
Traceback (most recent call last):
...
CancelledError
By default, shellous will send a SIGTERM signal to the process to tell it to exit. If the process does not
exit within 3 seconds, shellous will send a SIGKILL signal. You can change these defaults with the
cancel_signal and cancel_timeout settings. A command is not considered fully cancelled until the
process exits.
Pseudo-Terminal Support (Unix Only)
To run a command through a pseudo-terminal, use the pty adapter.
>>> await sh.pty("echo", "in a pty")
'in a pty\r\n'
Alternatively, you can pass a pty function to configure the tty mode and size.
>>> ls = sh("ls").set(pty=shellous.cooked(cols=40, rows=10, echo=False))
>>> await ls("README.md", "CHANGELOG.md")
'CHANGELOG.md\tREADME.md\r\n'
Shellous provides three built-in helper functions: shellous.cooked(), shellous.raw() and shellous.cbreak().
Context Objects
You can store shared command settings in an immutable context object (CmdContext). To create a new context object, specify your changes to the default context sh:
>>> auditor = lambda phase, info: print(phase, info["runner"].name)
>>> sh_audit = sh.set(audit_callback=auditor)
Now all commands created with sh_audit will log their progress using the audit callback.
>>> await sh_audit("echo", "goodbye")
start echo
stop echo
'goodbye\n'
You can also create a context object that specifies all return values are Result objects.
>>> rsh = sh.result
>>> await rsh("echo", "whatever")
Result(exit_code=0, output_bytes=b'whatever\n', error_bytes=b'', cancelled=False, encoding='utf-8')
Options
Both Command and CmdContext support options to control their runtime behavior. Some of these options (timeout,
pty, audit_callback, and exit_codes) have been described above. See the shellous.Options class for
more information.
You can retrieve an option from cmd with cmd.options.<option>. For example, use cmd.options.encoding to obtain the encoding:
>>> cmd = sh("echo").set(encoding="latin1")
>>> cmd.options.encoding
'latin1'
Command and CmdContext use the .set() method to specify most options:
| Option | Description |
|---|---|
| path | Search path to use instead of the PATH environment variable. (Default=None) |
| env | Additional environment variables to pass to the command. (Default={}) |
| inherit_env | True if command should inherit the environment variables from the current process. (Default=True) |
| encoding | Text encoding of input/output streams. You can specify an error handling scheme by including it after a space, e.g. "ascii backslashreplace". (Default="utf-8 strict") |
| exit_codes | Set of exit codes that do not raise a ResultError. (Default={0}) |
| timeout | Timeout in seconds to wait before cancelling the process. (Default=None) |
| cancel_timeout | Timeout in seconds to wait for a cancelled process to exit before forcefully terminating it. (Default=3s) |
| cancel_signal | The signal sent to a process when it is cancelled. (Default=SIGTERM) |
| alt_name | Alternate name for the process used for debug logging. (Default=None) |
| pass_fds | Additional file descriptors to pass to the process. (Default={}) |
| pass_fds_close | True if descriptors in pass_fds should be closed after the child process is launched. (Default=False) |
| pty | Used to allocate a pseudo-terminal (PTY). (Default=False) |
| close_fds | True if process should close all file descriptors when it starts. This setting defaults to False to align with posix_spawn requirements. (Default=False) |
| audit_callback | Provide function to audit stages of process execution. (Default=None) |
| coerce_arg | Provide function to coerce Command arguments to strings when str() is not sufficient. For example, you can provide your own function that converts a dictionary argument to a sequence of strings. (Default=None) |
| error_limit | Maximum number of initial bytes of STDERR to store in Result object. (Default=1024) |
| read_buffer_limit | Maximum number of bytes to read when looking for a separator. (Default=65536) |
env
Use the env() method to add to the list of environment variables. The env() method supports keyword parameters.
You can call env() more than once and the effect is additive.
>>> cmd = sh("echo").env(ENV1="a", ENV2="b").env(ENV2=3)
>>> cmd.options.env
{'ENV1': 'a', 'ENV2': '3'}
Use the env option with set() when you want to replace all the environment variables.
input, output, error
When you apply a redirection operator to a Command or CmdContext, the redirection targets
are also stored in the Options object. To change these, use the .stdin(), .stdout(), or .stderr() methods or the redirection operator |.
| Option | Description |
|---|---|
| input | The redirection target for standard input. |
| input_close | True if standard input should be closed after the process is launched. |
| output | The redirection target for standard output. |
| output_append | True if standard output should be open for append. |
| output_close | True if standard output should be closed after the process is launched. |
| error | The redirection target for standard error. |
| error_append | True if standard error should be open for append. |
| error_close | True if standard error should be closed after the process is launched. |
Type Checking
Shellous fully supports PEP 484 type hints.
Commands
Commands are generic on the return type, either str or Result. You will specify the
type of a command object as Command[str] or Command[Result].
Use the result modifier to obtain a Command[Result] from a Command[str].
from shellous import sh, Command, Result
cmd1: Command[str] = sh("echo", "abc")
# When you `await cmd1`, the result is a `str` object.
cmd2: Command[Result] = sh.result("echo", "abc")
# When you `await cmd2`, the result is a `Result` object.
CmdContext
The CmdContext class is also generic on either str or Result.
from shellous import sh, CmdContext, Result
sh1: CmdContext[str] = sh.set(path="/bin:/usr/bin")
# When you use `sh1` to create commands, it produces `Command[str]` object with the given path.
sh2: CmdContext[Result] = sh.result.set(path="/bin:/usr/bin")
# When you use `sh2` to create commands, it produces `Command[Result]` objects with the given path.
Logging
For verbose logging, shellous supports a SHELLOUS_TRACE environment variable. Set the
value of SHELLOUS_TRACE to a comma-delimited list of options:
detail: Enables detailed logging used to trace the steps of running a command.
prompt: Enables logging in the
Promptclass when controlling a program using send/expect.all: Enables all logging options.
Shellous uses the built-in Python logging module. After enabling these options,
the shellous logger will display log messages at the INFO level.
Without these options enabled, Shellous generates almost no log messages.
-
If you use an async generator object for
stdinorstdout, the command cannot run more than once. In Python, async ↩
1""" 2.. include:: ../README.md 3""" 4 5# pylint: disable=cyclic-import 6# pyright: reportUnusedImport=false 7 8__version__ = "0.41.0" 9 10import sys 11import warnings 12 13from .command import AuditEventInfo, CmdContext, Command, Options 14from .pipeline import Pipeline 15from .prompt import Prompt 16from .pty_util import cbreak, cooked, raw 17from .result import Result, ResultError 18from .runner import PipeRunner, Runner 19 20if sys.version_info[:3] in [(3, 10, 9), (3, 11, 1)]: 21 # Warn about these specific Python releases: 3.10.9 and 3.11.1 22 # These releases have a known race condition. 23 warnings.warn( # pragma: no cover 24 "Python 3.10.9 and Python 3.11.1 are unreliable with respect to " 25 + "asyncio subprocesses. Consider a newer Python release: 3.10.10+ " 26 + "or 3.11.2+. (https://github.com/python/cpython/issues/100133)", 27 RuntimeWarning, 28 ) 29 30 31sh: CmdContext[str] = CmdContext() 32"""`sh` is the default command context (`CmdContext`). 33 34Use `sh` to create commands or new command contexts. 35 36```python 37from shellous import sh 38result = await sh("echo", "hello") 39``` 40""" 41 42__all__ = [ 43 "sh", 44 "CmdContext", 45 "Command", 46 "Options", 47 "Pipeline", 48 "Prompt", 49 "cbreak", 50 "cooked", 51 "raw", 52 "Result", 53 "ResultError", 54 "Runner", 55 "PipeRunner", 56 "AuditEventInfo", 57]
sh: CmdContext[str] =
CmdContext(options=Options(path=None, inherit_env=True, input=<Redirect.DEFAULT: -20>, input_close=False, output=<Redirect.DEFAULT: -20>, output_append=False, output_close=False, error=<Redirect.DEFAULT: -20>, error_append=False, error_close=False, error_limit=1024, encoding='utf-8', _return_result=False, _catch_cancelled_error=False, exit_codes=None, timeout=None, cancel_timeout=3.0, cancel_signal=<Signals.SIGTERM: 15>, alt_name=None, pass_fds=(), pass_fds_close=False, _writable=False, _start_new_session=False, _preexec_fn=None, pty=False, close_fds=True, audit_callback=None, coerce_arg=None, read_buffer_limit=None))
sh is the default command context (CmdContext).
Use sh to create commands or new command contexts.
from shellous import sh
result = await sh("echo", "hello")
@dataclass(frozen=True)
class
CmdContext285@dataclass(frozen=True) 286class CmdContext(Generic[_RT]): 287 """Concrete class for an immutable execution context.""" 288 289 CAPTURE: ClassVar[Redirect] = Redirect.CAPTURE 290 "Capture and read/write stream manually." 291 292 DEVNULL: ClassVar[Redirect] = Redirect.DEVNULL 293 "Redirect to /dev/null." 294 295 INHERIT: ClassVar[Redirect] = Redirect.INHERIT 296 "Redirect to same place as existing stdin/stderr/stderr." 297 298 STDOUT: ClassVar[Redirect] = Redirect.STDOUT 299 "Redirect stderr to same place as stdout." 300 301 BUFFER: ClassVar[Redirect] = Redirect.BUFFER 302 "Redirect output to a buffer in the Result object. This is the default for stdout/stderr." 303 304 options: Options = field(default_factory=Options) 305 "Default command options." 306 307 def stdin( 308 self, 309 input_: Any, 310 *, 311 close: bool = False, 312 ) -> "CmdContext[_RT]": 313 "Return new context with updated `input` settings." 314 new_options = self.options.set_stdin(input_, close) 315 return CmdContext(new_options) 316 317 def stdout( 318 self, 319 output: Any, 320 *, 321 append: bool = False, 322 close: bool = False, 323 ) -> "CmdContext[_RT]": 324 "Return new context with updated `output` settings." 325 new_options = self.options.set_stdout(output, append, close) 326 return CmdContext(new_options) 327 328 def stderr( 329 self, 330 error: Any, 331 *, 332 append: bool = False, 333 close: bool = False, 334 ) -> "CmdContext[_RT]": 335 "Return new context with updated `error` settings." 336 new_options = self.options.set_stderr(error, append, close) 337 return CmdContext(new_options) 338 339 def env(self, **kwds: Any) -> "CmdContext[_RT]": 340 """Return new context with augmented environment.""" 341 new_options = self.options.add_env(kwds) 342 return CmdContext(new_options) 343 344 def set( # pylint: disable=unused-argument, too-many-locals, too-many-arguments 345 self, 346 *, 347 path: Unset[str | None] = _UNSET, 348 env: Unset[dict[str, Any]] = _UNSET, 349 inherit_env: Unset[bool] = _UNSET, 350 encoding: Unset[str] = _UNSET, 351 _return_result: Unset[bool] = _UNSET, 352 _catch_cancelled_error: Unset[bool] = _UNSET, 353 exit_codes: Unset[Container[int] | None] = _UNSET, 354 timeout: Unset[float | None] = _UNSET, 355 cancel_timeout: Unset[float] = _UNSET, 356 cancel_signal: Unset[signal.Signals | None] = _UNSET, 357 alt_name: Unset[str | None] = _UNSET, 358 pass_fds: Unset[Iterable[int]] = _UNSET, 359 pass_fds_close: Unset[bool] = _UNSET, 360 _writable: Unset[bool] = _UNSET, 361 _start_new_session: Unset[bool] = _UNSET, 362 _preexec_fn: Unset[_PreexecFnT] = _UNSET, 363 pty: Unset[PtyAdapterOrBool] = _UNSET, 364 close_fds: Unset[bool] = _UNSET, 365 audit_callback: Unset[_AuditFnT] = _UNSET, 366 coerce_arg: Unset[_CoerceArgFnT] = _UNSET, 367 error_limit: Unset[int | None] = _UNSET, 368 ) -> "CmdContext[_RT]": 369 """Return new context with custom options set. 370 371 See `Command.set` for option reference. 372 """ 373 kwargs = locals() 374 del kwargs["self"] 375 if not encoding: 376 raise TypeError("invalid encoding") 377 return CmdContext(self.options.set(kwargs)) 378 379 def __call__(self, *args: Any) -> "Command[_RT]": 380 "Construct a new command." 381 return Command(coerce(args, self.options.coerce_arg), self.options) 382 383 @property 384 def pty(self) -> "CmdContext[_RT]": 385 "Set `pty` to true." 386 return self.set(pty=True) 387 388 @property 389 def result(self) -> "CmdContext[shellous.Result]": 390 "Set `_return_result` and `exit_codes`." 391 return cast( 392 CmdContext[shellous.Result], 393 self.set( 394 _return_result=True, 395 exit_codes=range(-255, 2**32), 396 ), 397 ) 398 399 def find_command(self, name: str) -> Path | None: 400 """Find the command with the given name and return its filesystem path. 401 402 Return None if the command name is not found in the search path. 403 404 Use the `path` variable specified by the context if set. Otherwise, the 405 default behavior is to use the `PATH` environment variable with a 406 fallback to the value of `os.defpath`. 407 """ 408 result = self.options.which(name) 409 if not result: 410 return None 411 return Path(result)
Concrete class for an immutable execution context.
CmdContext(options: Options = <factory>)
CAPTURE: ClassVar[shellous.redirect.Redirect] =
<Redirect.CAPTURE: -10>
Capture and read/write stream manually.
INHERIT: ClassVar[shellous.redirect.Redirect] =
<Redirect.INHERIT: -11>
Redirect to same place as existing stdin/stderr/stderr.
STDOUT: ClassVar[shellous.redirect.Redirect] =
<Redirect.STDOUT: -2>
Redirect stderr to same place as stdout.
BUFFER: ClassVar[shellous.redirect.Redirect] =
<Redirect.BUFFER: -12>
Redirect output to a buffer in the Result object. This is the default for stdout/stderr.
307 def stdin( 308 self, 309 input_: Any, 310 *, 311 close: bool = False, 312 ) -> "CmdContext[_RT]": 313 "Return new context with updated `input` settings." 314 new_options = self.options.set_stdin(input_, close) 315 return CmdContext(new_options)
Return new context with updated input settings.
317 def stdout( 318 self, 319 output: Any, 320 *, 321 append: bool = False, 322 close: bool = False, 323 ) -> "CmdContext[_RT]": 324 "Return new context with updated `output` settings." 325 new_options = self.options.set_stdout(output, append, close) 326 return CmdContext(new_options)
Return new context with updated output settings.
328 def stderr( 329 self, 330 error: Any, 331 *, 332 append: bool = False, 333 close: bool = False, 334 ) -> "CmdContext[_RT]": 335 "Return new context with updated `error` settings." 336 new_options = self.options.set_stderr(error, append, close) 337 return CmdContext(new_options)
Return new context with updated error settings.
339 def env(self, **kwds: Any) -> "CmdContext[_RT]": 340 """Return new context with augmented environment.""" 341 new_options = self.options.add_env(kwds) 342 return CmdContext(new_options)
Return new context with augmented environment.
def
set( self, *, path: Union[str, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, env: Union[dict[str, Any], shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, inherit_env: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, encoding: Union[str, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _return_result: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _catch_cancelled_error: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, exit_codes: Union[Container[int], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, timeout: Union[float, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, cancel_timeout: Union[float, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, cancel_signal: Union[signal.Signals, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, alt_name: Union[str, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, pass_fds: Union[Iterable[int], shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, pass_fds_close: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _writable: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _start_new_session: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _preexec_fn: Union[Callable[[], NoneType], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, pty: Union[Callable[[int], NoneType], bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, close_fds: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, audit_callback: Union[Callable[[str, AuditEventInfo], NoneType], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, coerce_arg: Union[Callable[[Any], Any], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, error_limit: Union[int, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>) -> CmdContext[~_RT]:
344 def set( # pylint: disable=unused-argument, too-many-locals, too-many-arguments 345 self, 346 *, 347 path: Unset[str | None] = _UNSET, 348 env: Unset[dict[str, Any]] = _UNSET, 349 inherit_env: Unset[bool] = _UNSET, 350 encoding: Unset[str] = _UNSET, 351 _return_result: Unset[bool] = _UNSET, 352 _catch_cancelled_error: Unset[bool] = _UNSET, 353 exit_codes: Unset[Container[int] | None] = _UNSET, 354 timeout: Unset[float | None] = _UNSET, 355 cancel_timeout: Unset[float] = _UNSET, 356 cancel_signal: Unset[signal.Signals | None] = _UNSET, 357 alt_name: Unset[str | None] = _UNSET, 358 pass_fds: Unset[Iterable[int]] = _UNSET, 359 pass_fds_close: Unset[bool] = _UNSET, 360 _writable: Unset[bool] = _UNSET, 361 _start_new_session: Unset[bool] = _UNSET, 362 _preexec_fn: Unset[_PreexecFnT] = _UNSET, 363 pty: Unset[PtyAdapterOrBool] = _UNSET, 364 close_fds: Unset[bool] = _UNSET, 365 audit_callback: Unset[_AuditFnT] = _UNSET, 366 coerce_arg: Unset[_CoerceArgFnT] = _UNSET, 367 error_limit: Unset[int | None] = _UNSET, 368 ) -> "CmdContext[_RT]": 369 """Return new context with custom options set. 370 371 See `Command.set` for option reference. 372 """ 373 kwargs = locals() 374 del kwargs["self"] 375 if not encoding: 376 raise TypeError("invalid encoding") 377 return CmdContext(self.options.set(kwargs))
Return new context with custom options set.
See Command.set for option reference.
pty: CmdContext[~_RT]
383 @property 384 def pty(self) -> "CmdContext[_RT]": 385 "Set `pty` to true." 386 return self.set(pty=True)
Set pty to true.
result: CmdContext[Result]
388 @property 389 def result(self) -> "CmdContext[shellous.Result]": 390 "Set `_return_result` and `exit_codes`." 391 return cast( 392 CmdContext[shellous.Result], 393 self.set( 394 _return_result=True, 395 exit_codes=range(-255, 2**32), 396 ), 397 )
Set _return_result and exit_codes.
def
find_command(self, name: str) -> pathlib.Path | None:
399 def find_command(self, name: str) -> Path | None: 400 """Find the command with the given name and return its filesystem path. 401 402 Return None if the command name is not found in the search path. 403 404 Use the `path` variable specified by the context if set. Otherwise, the 405 default behavior is to use the `PATH` environment variable with a 406 fallback to the value of `os.defpath`. 407 """ 408 result = self.options.which(name) 409 if not result: 410 return None 411 return Path(result)
Find the command with the given name and return its filesystem path.
Return None if the command name is not found in the search path.
Use the path variable specified by the context if set. Otherwise, the
default behavior is to use the PATH environment variable with a
fallback to the value of os.defpath.
@dataclass(frozen=True)
class
Command414@dataclass(frozen=True) 415class Command(Generic[_RT]): 416 """A Command instance is lightweight and immutable object that specifies the 417 arguments and options used to run a program. Commands do not do anything 418 until they are awaited. 419 420 Commands are always created by a CmdContext. 421 422 ``` 423 from shellous import sh 424 425 # Create a new command from the context. 426 echo = sh("echo", "hello, world") 427 428 # Run the command. 429 result = await echo 430 ``` 431 """ 432 433 args: "tuple[str | bytes | os.PathLike[Any] | Command[Any] | shellous.Pipeline[Any], ...]" 434 "Command arguments including the program name as first argument." 435 436 options: Options 437 "Command options." 438 439 def __post_init__(self) -> None: 440 "Validate the command." 441 if len(self.args) == 0: 442 raise ValueError("Command must include program name") 443 444 @property 445 def name(self) -> str: 446 """Returns the name of the program being run. 447 448 Names longer than 31 characters are truncated. If `alt_name` option 449 is set, return that instead. 450 """ 451 if self.options.alt_name: 452 return self.options.alt_name 453 name = str(self.args[0]) 454 if len(name) > 31: 455 return f"...{name[-31:]}" 456 return name 457 458 def stdin(self, input_: Any, *, close: bool = False) -> "Command[_RT]": 459 "Pass `input` to command's standard input." 460 new_options = self.options.set_stdin(input_, close) 461 return Command(self.args, new_options) 462 463 def stdout( 464 self, 465 output: Any, 466 *, 467 append: bool = False, 468 close: bool = False, 469 ) -> "Command[_RT]": 470 "Redirect standard output to `output`." 471 new_options = self.options.set_stdout(output, append, close) 472 return Command(self.args, new_options) 473 474 def stderr( 475 self, 476 error: Any, 477 *, 478 append: bool = False, 479 close: bool = False, 480 ) -> "Command[_RT]": 481 "Redirect standard error to `error`." 482 new_options = self.options.set_stderr(error, append, close) 483 return Command(self.args, new_options) 484 485 def env(self, **kwds: Any) -> "Command[_RT]": 486 """Return new command with augmented environment. 487 488 The changes to the environment variables made by this method are 489 additive. For example, calling `cmd.env(A=1).env(B=2)` produces a 490 command with the environment set to `{"A": "1", "B": "2"}`. 491 492 To clear the environment, use the `cmd.set(env={})` method. 493 """ 494 new_options = self.options.add_env(kwds) 495 return Command(self.args, new_options) 496 497 def set( # pylint: disable=unused-argument, too-many-locals, too-many-arguments 498 self, 499 *, 500 path: Unset[str | None] = _UNSET, 501 env: Unset[dict[str, Any]] = _UNSET, 502 inherit_env: Unset[bool] = _UNSET, 503 encoding: Unset[str] = _UNSET, 504 _return_result: Unset[bool] = _UNSET, 505 _catch_cancelled_error: Unset[bool] = _UNSET, 506 exit_codes: Unset[Container[int] | None] = _UNSET, 507 timeout: Unset[float | None] = _UNSET, 508 cancel_timeout: Unset[float] = _UNSET, 509 cancel_signal: Unset[signal.Signals | None] = _UNSET, 510 alt_name: Unset[str | None] = _UNSET, 511 pass_fds: Unset[Iterable[int]] = _UNSET, 512 pass_fds_close: Unset[bool] = _UNSET, 513 _writable: Unset[bool] = _UNSET, 514 _start_new_session: Unset[bool] = _UNSET, 515 _preexec_fn: Unset[_PreexecFnT] = _UNSET, 516 pty: Unset[PtyAdapterOrBool] = _UNSET, 517 close_fds: Unset[bool] = _UNSET, 518 audit_callback: Unset[_AuditFnT] = _UNSET, 519 coerce_arg: Unset[_CoerceArgFnT] = _UNSET, 520 error_limit: Unset[int | None] = _UNSET, 521 read_buffer_limit: Unset[int | None] = _UNSET, 522 ) -> "Command[_RT]": 523 """Return new command with custom options set. 524 525 **path** (str | None) default=None<br> 526 Search path for locating command executable. By default, `path` is None 527 which causes shellous to rely on the `PATH` environment variable. 528 529 **env** (dict[str, str]) default={}<br> 530 Set the environment variables for the subprocess. If `inherit_env` is 531 True, the subprocess will also inherit the environment variables 532 specified by the parent process. 533 534 Using `set(env=...)` will replace all environment variables using the 535 dictionary argument. You can also use the `env(...)` method to modify 536 the existing environment incrementally. 537 538 **inherit_env** (bool) default=True<br> 539 Subprocess should inherit the parent process environment. If this is 540 False, the subprocess will only have environment variables specified 541 by `Command.env`. If `inherit_env` is True, the parent process 542 environment is augmented/overridden by any variables specified in 543 `Command.env`. 544 545 **encoding** (str) default="utf-8"<br> 546 String encoding to use for subprocess input/output. To specify `errors`, 547 append it after a space. For example, use "utf-8 replace" to specify 548 "utf-8" with errors "replace". 549 550 **_return_result** (bool) default=False<br> 551 When True, return a `Result` object instead of the standard output. 552 Private API -- use the `result` modifier instead. 553 554 **_catch_cancelled_error** (bool) default=False<br> 555 When True, raise a `ResultError` when the command is cancelled. 556 Private API -- used internally by PipeRunner. 557 558 **exit_codes** (set[int] | None) default=None<br> 559 Set of allowed exit codes that will not raise a `ResultError`. By default, 560 `exit_codes` is `None` which indicates that 0 is the only valid exit 561 status. Any other exit status will raise a `ResultError`. In addition to 562 sets of integers, you can use a `range` object, e.g. `range(256)` for 563 any positive exit status. 564 565 **timeout** (float | None) default=None<br> 566 Timeout in seconds to wait before we cancel the process. The timer 567 begins immediately after the process is launched. This differs from 568 using `asyncio.wait_for` which includes the process launch time also. 569 If timeout is None (the default), there is no timeout. 570 571 **cancel_timeout** (float) default=3.0 seconds<br> 572 Timeout in seconds to wait for a process to exit after sending it a 573 `cancel_signal`. If the process does not exit after waiting for 574 `cancel_timeout` seconds, we send a kill signal to the process. 575 576 **cancel_signal** (signals.Signal | None) default=signal.SIGTERM<br> 577 Signal sent to a process when it is cancelled. If `cancel_signal` is 578 None, send a `SIGKILL` on Unix and `SIGTERM` (TerminateProcess) on 579 Windows. 580 581 **alt_name** (str| None) default=None<br> 582 Alternative name of the command displayed in logs. Used to resolve 583 ambiguity when the actual command name is a scripting language. 584 585 **pass_fds** (Iterable[int]) default=()<br> 586 Specify open file descriptors to pass to the subprocess. 587 588 **pass_fds_close** (bool) default=False<br> 589 Close the file descriptors in `pass_fds` immediately in the current 590 process immediately after launching the subprocess. 591 592 **_writable** (bool) default=False<br> 593 Used to indicate process substitution is writing. 594 Private API -- use the `writable` modifier instead. 595 596 **_start_new_session** (bool) default=False<br> 597 Private API -- provided for testing purposes only. 598 599 **_preexec_fn** (Callable() | None) default=None<br> 600 Private API -- provided for testing purposes only. 601 602 **pty** (bool | Callable(int)) default=False<br> 603 If True, use a pseudo-terminal (pty) to control the child process. 604 If `pty` is set to a callable, the function must take one int argument 605 for the child side of the pty. The function is called to set the child 606 pty's termios settings before spawning the subprocess. 607 608 shellous provides three utility functions: `shellous.cooked`, 609 `shellous.raw` and `shellous.cbreak` that can be used as arguments to 610 the `pty` option. 611 612 **close_fds** (bool) default=True<br> 613 Close all unnecessary file descriptors in the child process. This 614 defaults to True to align with the default behavior of the subprocess 615 module. 616 617 **audit_callback** (Callable(phase, info) | None) default=None<br> 618 Specify a function to call as the command execution goes through its 619 lifecycle. `audit_callback` is a function called with two arguments, 620 *phase* and *info*. 621 622 *phase* can be one of three values: 623 624 "start": The process is about to start. 625 626 "stop": The process finally stopped. 627 628 "signal": The process is being sent a signal. 629 630 *info* is a dictionary providing more information for the callback. The 631 following keys are currently defined: 632 633 "runner" (Runner): Reference to the Runner object. 634 635 "failure" (str): When phase is "stop", optional string with the 636 name of the exception from launching the process. 637 638 "signal" (str): When phase is "signal", the signal name/number 639 sent to the process, e.g. "SIGTERM". 640 641 The primary use case for `audit_callback` is measuring how long each 642 command takes to run and exporting this information to a metrics 643 framework like Prometheus. 644 645 **coerce_arg** (Callable(arg) | None) default=None<br> 646 Specify a function to call on each command line argument. This function 647 can specify how to coerce unsupported argument types (e.g. dict) to 648 a sequence of strings. This function should return the original value 649 unchanged if there is no conversion needed. 650 651 **error_limit** (int | None) default=1024<br> 652 Specify the number of bytes to store when redirecting STDERR to BUFFER. 653 After reading up to `error_limit` bytes, shellous will continue to 654 read from stderr, but will not store any additional bytes. Setting 655 `error_limit` only affects the internal BUFFER; it has no effect when 656 using other redirection types. 657 658 **read_buffer_limit** (int | None) default=65536<br> 659 Specify the maximum number of bytes to read from a stdout/stderr stream 660 when looking for a separator. Increase this value if you expect the 661 subprocess to emit extremely long lines. If this value is too small, 662 you may get the error: "Separator is not found". 663 664 """ 665 kwargs = locals() 666 del kwargs["self"] 667 if not encoding: 668 raise TypeError("invalid encoding") 669 return Command(self.args, self.options.set(kwargs)) 670 671 def _replace_args(self, new_args: Sequence[Any]) -> "Command[_RT]": 672 """Return new command with arguments replaced by `new_args`. 673 674 Arguments are NOT type-checked by the context. Program name must be the 675 exact same object. 676 """ 677 assert new_args 678 assert new_args[0] is self.args[0] 679 return Command(tuple(new_args), self.options) 680 681 def coro( 682 self, 683 *, 684 _run_future: asyncio.Future[Runner] | None = None, 685 ) -> Coroutine[Any, Any, _RT]: 686 "Return coroutine object to run awaitable." 687 return cast( 688 Coroutine[Any, Any, _RT], 689 Runner.run_command(self, _run_future=_run_future), 690 ) 691 692 @contextlib.asynccontextmanager 693 async def prompt( 694 self, 695 prompt: str | list[str] | re.Pattern[str] | None = None, 696 *, 697 timeout: float | None = None, 698 normalize_newlines: bool = False, 699 ) -> AsyncGenerator[Prompt, None]: 700 """Run command using the send/expect API. 701 702 This method should be called using `async with`. It returns a `Prompt` 703 object with send() and expect() methods. 704 705 You can optionally set a default `prompt`. This is used by `expect()` 706 when you don't provide another value. 707 708 Use the `timeout` parameter to set the default timeout for operations. 709 710 Set `normalize_newlines` to True to convert incoming CR and CR-LF to LF. 711 This conversion is done before matching with `expect()`. This option 712 does not affect strings sent with `send()`. 713 """ 714 cmd = self.stdin(Redirect.CAPTURE).stdout(Redirect.CAPTURE) 715 716 cli = None 717 try: 718 async with Runner(cmd) as run: 719 cli = Prompt( 720 run, 721 default_prompt=prompt, 722 default_timeout=timeout, 723 normalize_newlines=normalize_newlines, 724 ) 725 yield cli 726 cli.close() 727 finally: 728 if cli is not None: 729 cli._finish_() # pyright: ignore[reportPrivateUsage] 730 731 def __await__(self) -> "Generator[Any, None, _RT]": 732 "Run process and return the standard output." 733 return self.coro().__await__() 734 735 async def __aenter__(self) -> Runner: 736 "Enter the async context manager." 737 return await context_aenter(self, Runner(self)) 738 739 async def __aexit__( 740 self, 741 exc_type: type[BaseException] | None, 742 exc_value: BaseException | None, 743 exc_tb: TracebackType | None, 744 ) -> bool | None: 745 "Exit the async context manager." 746 return await context_aexit(self, exc_type, exc_value, exc_tb) 747 748 def __aiter__(self) -> AsyncIterator[str]: 749 "Return async iterator to iterate over output lines." 750 return aiter_preflight(self)._readlines() 751 752 async def _readlines(self) -> AsyncIterator[str]: 753 "Async generator to iterate over lines." 754 async with Runner(self) as run: 755 async for line in run: 756 yield line 757 758 def __call__(self, *args: Any) -> "Command[_RT]": 759 "Apply more arguments to the end of the command." 760 if not args: 761 return self 762 new_args = self.args + coerce(args, self.options.coerce_arg) 763 return Command(new_args, self.options) 764 765 def __str__(self) -> str: 766 """Return string representation for command. 767 768 Display the full name of the command only. Don't include arguments or 769 environment variables. 770 """ 771 return str(self.args[0]) 772 773 @overload 774 def __or__(self, rhs: StdoutType) -> "Command[_RT]": ... # pragma: no cover 775 776 @overload 777 def __or__( 778 self, rhs: "Command[str]" 779 ) -> "shellous.Pipeline[str]": ... # pragma: no cover 780 781 @overload 782 def __or__( 783 self, 784 rhs: "Command[shellous.Result]", 785 ) -> "shellous.Pipeline[shellous.Result]": ... # pragma: no cover 786 787 def __or__(self, rhs: Any) -> Any: 788 "Bitwise or operator is used to build pipelines." 789 if isinstance(rhs, STDOUT_TYPES): 790 return self.stdout(rhs) 791 return shellous.Pipeline.create(self) | rhs 792 793 def __ror__(self, lhs: StdinType) -> "Command[_RT]": 794 "Bitwise or operator is used to build pipelines." 795 if isinstance(lhs, STDIN_TYPES): # pyright: ignore[reportUnnecessaryIsInstance] 796 return self.stdin(lhs) 797 return NotImplemented 798 799 def __rshift__(self, rhs: StdoutType) -> "Command[_RT]": 800 "Right shift operator is used to build pipelines." 801 if isinstance( 802 rhs, STDOUT_TYPES 803 ): # pyright: ignore[reportUnnecessaryIsInstance] 804 return self.stdout(rhs, append=True) 805 return NotImplemented 806 807 @property 808 def writable(self) -> "Command[_RT]": 809 "Set `writable` to True." 810 return self.set(_writable=True) 811 812 @property 813 def pty(self) -> "Command[_RT]": 814 "Set `pty` to True." 815 return self.set(pty=True) 816 817 @property 818 def result(self) -> "Command[shellous.Result]": 819 "Set `_return_result` and `exit_codes`." 820 return cast( 821 Command[shellous.Result], 822 self.set(_return_result=True, exit_codes=range(-255, 256)), 823 )
A Command instance is lightweight and immutable object that specifies the arguments and options used to run a program. Commands do not do anything until they are awaited.
Commands are always created by a CmdContext.
from shellous import sh
# Create a new command from the context.
echo = sh("echo", "hello, world")
# Run the command.
result = await echo
Command( args: tuple[typing.Union[str, bytes, os.PathLike[typing.Any], Command[typing.Any], Pipeline[typing.Any]], ...], options: Options)
args: tuple[typing.Union[str, bytes, os.PathLike[typing.Any], Command[typing.Any], Pipeline[typing.Any]], ...]
Command arguments including the program name as first argument.
name: str
444 @property 445 def name(self) -> str: 446 """Returns the name of the program being run. 447 448 Names longer than 31 characters are truncated. If `alt_name` option 449 is set, return that instead. 450 """ 451 if self.options.alt_name: 452 return self.options.alt_name 453 name = str(self.args[0]) 454 if len(name) > 31: 455 return f"...{name[-31:]}" 456 return name
Returns the name of the program being run.
Names longer than 31 characters are truncated. If alt_name option
is set, return that instead.
458 def stdin(self, input_: Any, *, close: bool = False) -> "Command[_RT]": 459 "Pass `input` to command's standard input." 460 new_options = self.options.set_stdin(input_, close) 461 return Command(self.args, new_options)
Pass input to command's standard input.
463 def stdout( 464 self, 465 output: Any, 466 *, 467 append: bool = False, 468 close: bool = False, 469 ) -> "Command[_RT]": 470 "Redirect standard output to `output`." 471 new_options = self.options.set_stdout(output, append, close) 472 return Command(self.args, new_options)
Redirect standard output to output.
474 def stderr( 475 self, 476 error: Any, 477 *, 478 append: bool = False, 479 close: bool = False, 480 ) -> "Command[_RT]": 481 "Redirect standard error to `error`." 482 new_options = self.options.set_stderr(error, append, close) 483 return Command(self.args, new_options)
Redirect standard error to error.
485 def env(self, **kwds: Any) -> "Command[_RT]": 486 """Return new command with augmented environment. 487 488 The changes to the environment variables made by this method are 489 additive. For example, calling `cmd.env(A=1).env(B=2)` produces a 490 command with the environment set to `{"A": "1", "B": "2"}`. 491 492 To clear the environment, use the `cmd.set(env={})` method. 493 """ 494 new_options = self.options.add_env(kwds) 495 return Command(self.args, new_options)
Return new command with augmented environment.
The changes to the environment variables made by this method are
additive. For example, calling cmd.env(A=1).env(B=2) produces a
command with the environment set to {"A": "1", "B": "2"}.
To clear the environment, use the cmd.set(env={}) method.
def
set( self, *, path: Union[str, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, env: Union[dict[str, Any], shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, inherit_env: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, encoding: Union[str, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _return_result: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _catch_cancelled_error: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, exit_codes: Union[Container[int], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, timeout: Union[float, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, cancel_timeout: Union[float, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, cancel_signal: Union[signal.Signals, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, alt_name: Union[str, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, pass_fds: Union[Iterable[int], shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, pass_fds_close: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _writable: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _start_new_session: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, _preexec_fn: Union[Callable[[], NoneType], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, pty: Union[Callable[[int], NoneType], bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, close_fds: Union[bool, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, audit_callback: Union[Callable[[str, AuditEventInfo], NoneType], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, coerce_arg: Union[Callable[[Any], Any], NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, error_limit: Union[int, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>, read_buffer_limit: Union[int, NoneType, shellous.command._UnsetEnum] = <_UnsetEnum.UNSET: 1>) -> Command[~_RT]:
497 def set( # pylint: disable=unused-argument, too-many-locals, too-many-arguments 498 self, 499 *, 500 path: Unset[str | None] = _UNSET, 501 env: Unset[dict[str, Any]] = _UNSET, 502 inherit_env: Unset[bool] = _UNSET, 503 encoding: Unset[str] = _UNSET, 504 _return_result: Unset[bool] = _UNSET, 505 _catch_cancelled_error: Unset[bool] = _UNSET, 506 exit_codes: Unset[Container[int] | None] = _UNSET, 507 timeout: Unset[float | None] = _UNSET, 508 cancel_timeout: Unset[float] = _UNSET, 509 cancel_signal: Unset[signal.Signals | None] = _UNSET, 510 alt_name: Unset[str | None] = _UNSET, 511 pass_fds: Unset[Iterable[int]] = _UNSET, 512 pass_fds_close: Unset[bool] = _UNSET, 513 _writable: Unset[bool] = _UNSET, 514 _start_new_session: Unset[bool] = _UNSET, 515 _preexec_fn: Unset[_PreexecFnT] = _UNSET, 516 pty: Unset[PtyAdapterOrBool] = _UNSET, 517 close_fds: Unset[bool] = _UNSET, 518 audit_callback: Unset[_AuditFnT] = _UNSET, 519 coerce_arg: Unset[_CoerceArgFnT] = _UNSET, 520 error_limit: Unset[int | None] = _UNSET, 521 read_buffer_limit: Unset[int | None] = _UNSET, 522 ) -> "Command[_RT]": 523 """Return new command with custom options set. 524 525 **path** (str | None) default=None<br> 526 Search path for locating command executable. By default, `path` is None 527 which causes shellous to rely on the `PATH` environment variable. 528 529 **env** (dict[str, str]) default={}<br> 530 Set the environment variables for the subprocess. If `inherit_env` is 531 True, the subprocess will also inherit the environment variables 532 specified by the parent process. 533 534 Using `set(env=...)` will replace all environment variables using the 535 dictionary argument. You can also use the `env(...)` method to modify 536 the existing environment incrementally. 537 538 **inherit_env** (bool) default=True<br> 539 Subprocess should inherit the parent process environment. If this is 540 False, the subprocess will only have environment variables specified 541 by `Command.env`. If `inherit_env` is True, the parent process 542 environment is augmented/overridden by any variables specified in 543 `Command.env`. 544 545 **encoding** (str) default="utf-8"<br> 546 String encoding to use for subprocess input/output. To specify `errors`, 547 append it after a space. For example, use "utf-8 replace" to specify 548 "utf-8" with errors "replace". 549 550 **_return_result** (bool) default=False<br> 551 When True, return a `Result` object instead of the standard output. 552 Private API -- use the `result` modifier instead. 553 554 **_catch_cancelled_error** (bool) default=False<br> 555 When True, raise a `ResultError` when the command is cancelled. 556 Private API -- used internally by PipeRunner. 557 558 **exit_codes** (set[int] | None) default=None<br> 559 Set of allowed exit codes that will not raise a `ResultError`. By default, 560 `exit_codes` is `None` which indicates that 0 is the only valid exit 561 status. Any other exit status will raise a `ResultError`. In addition to 562 sets of integers, you can use a `range` object, e.g. `range(256)` for 563 any positive exit status. 564 565 **timeout** (float | None) default=None<br> 566 Timeout in seconds to wait before we cancel the process. The timer 567 begins immediately after the process is launched. This differs from 568 using `asyncio.wait_for` which includes the process launch time also. 569 If timeout is None (the default), there is no timeout. 570 571 **cancel_timeout** (float) default=3.0 seconds<br> 572 Timeout in seconds to wait for a process to exit after sending it a 573 `cancel_signal`. If the process does not exit after waiting for 574 `cancel_timeout` seconds, we send a kill signal to the process. 575 576 **cancel_signal** (signals.Signal | None) default=signal.SIGTERM<br> 577 Signal sent to a process when it is cancelled. If `cancel_signal` is 578 None, send a `SIGKILL` on Unix and `SIGTERM` (TerminateProcess) on 579 Windows. 580 581 **alt_name** (str| None) default=None<br> 582 Alternative name of the command displayed in logs. Used to resolve 583 ambiguity when the actual command name is a scripting language. 584 585 **pass_fds** (Iterable[int]) default=()<br> 586 Specify open file descriptors to pass to the subprocess. 587 588 **pass_fds_close** (bool) default=False<br> 589 Close the file descriptors in `pass_fds` immediately in the current 590 process immediately after launching the subprocess. 591 592 **_writable** (bool) default=False<br> 593 Used to indicate process substitution is writing. 594 Private API -- use the `writable` modifier instead. 595 596 **_start_new_session** (bool) default=False<br> 597 Private API -- provided for testing purposes only. 598 599 **_preexec_fn** (Callable() | None) default=None<br> 600 Private API -- provided for testing purposes only. 601 602 **pty** (bool | Callable(int)) default=False<br> 603 If True, use a pseudo-terminal (pty) to control the child process. 604 If `pty` is set to a callable, the function must take one int argument 605 for the child side of the pty. The function is called to set the child 606 pty's termios settings before spawning the subprocess. 607 608 shellous provides three utility functions: `shellous.cooked`, 609 `shellous.raw` and `shellous.cbreak` that can be used as arguments to 610 the `pty` option. 611 612 **close_fds** (bool) default=True<br> 613 Close all unnecessary file descriptors in the child process. This 614 defaults to True to align with the default behavior of the subprocess 615 module. 616 617 **audit_callback** (Callable(phase, info) | None) default=None<br> 618 Specify a function to call as the command execution goes through its 619 lifecycle. `audit_callback` is a function called with two arguments, 620 *phase* and *info*. 621 622 *phase* can be one of three values: 623 624 "start": The process is about to start. 625 626 "stop": The process finally stopped. 627 628 "signal": The process is being sent a signal. 629 630 *info* is a dictionary providing more information for the callback. The 631 following keys are currently defined: 632 633 "runner" (Runner): Reference to the Runner object. 634 635 "failure" (str): When phase is "stop", optional string with the 636 name of the exception from launching the process. 637 638 "signal" (str): When phase is "signal", the signal name/number 639 sent to the process, e.g. "SIGTERM". 640 641 The primary use case for `audit_callback` is measuring how long each 642 command takes to run and exporting this information to a metrics 643 framework like Prometheus. 644 645 **coerce_arg** (Callable(arg) | None) default=None<br> 646 Specify a function to call on each command line argument. This function 647 can specify how to coerce unsupported argument types (e.g. dict) to 648 a sequence of strings. This function should return the original value 649 unchanged if there is no conversion needed. 650 651 **error_limit** (int | None) default=1024<br> 652 Specify the number of bytes to store when redirecting STDERR to BUFFER. 653 After reading up to `error_limit` bytes, shellous will continue to 654 read from stderr, but will not store any additional bytes. Setting 655 `error_limit` only affects the internal BUFFER; it has no effect when 656 using other redirection types. 657 658 **read_buffer_limit** (int | None) default=65536<br> 659 Specify the maximum number of bytes to read from a stdout/stderr stream 660 when looking for a separator. Increase this value if you expect the 661 subprocess to emit extremely long lines. If this value is too small, 662 you may get the error: "Separator is not found". 663 664 """ 665 kwargs = locals() 666 del kwargs["self"] 667 if not encoding: 668 raise TypeError("invalid encoding") 669 return Command(self.args, self.options.set(kwargs))
Return new command with custom options set.
path (str | None) default=None
Search path for locating command executable. By default, path is None
which causes shellous to rely on the PATH environment variable.
env (dict[str, str]) default={}
Set the environment variables for the subprocess. If inherit_env is
True, the subprocess will also inherit the environment variables
specified by the parent process.
Using set(env=...) will replace all environment variables using the
dictionary argument. You can also use the env(...) method to modify
the existing environment incrementally.
inherit_env (bool) default=True
Subprocess should inherit the parent process environment. If this is
False, the subprocess will only have environment variables specified
by Command.env. If inherit_env is True, the parent process
environment is augmented/overridden by any variables specified in
Command.env.
encoding (str) default="utf-8"
String encoding to use for subprocess input/output. To specify errors,
append it after a space. For example, use "utf-8 replace" to specify
"utf-8" with errors "replace".
_return_result (bool) default=False
When True, return a Result object instead of the standard output.
Private API -- use the result modifier instead.
_catch_cancelled_error (bool) default=False
When True, raise a ResultError when the command is cancelled.
Private API -- used internally by PipeRunner.
exit_codes (set[int] | None) default=None
Set of allowed exit codes that will not raise a ResultError. By default,
exit_codes is None which indicates that 0 is the only valid exit
status. Any other exit status will raise a ResultError. In addition to
sets of integers, you can use a range object, e.g. range(256) for
any positive exit status.
timeout (float | None) default=None
Timeout in seconds to wait before we cancel the process. The timer
begins immediately after the process is launched. This differs from
using asyncio.wait_for which includes the process launch time also.
If timeout is None (the default), there is no timeout.
cancel_timeout (float) default=3.0 seconds
Timeout in seconds to wait for a process to exit after sending it a
cancel_signal. If the process does not exit after waiting for
cancel_timeout seconds, we send a kill signal to the process.
cancel_signal (signals.Signal | None) default=signal.SIGTERM
Signal sent to a process when it is cancelled. If cancel_signal is
None, send a SIGKILL on Unix and SIGTERM (TerminateProcess) on
Windows.
alt_name (str| None) default=None
Alternative name of the command displayed in logs. Used to resolve
ambiguity when the actual command name is a scripting language.
pass_fds (Iterable[int]) default=()
Specify open file descriptors to pass to the subprocess.
pass_fds_close (bool) default=False
Close the file descriptors in pass_fds immediately in the current
process immediately after launching the subprocess.
_writable (bool) default=False
Used to indicate process substitution is writing.
Private API -- use the writable modifier instead.
_start_new_session (bool) default=False
Private API -- provided for testing purposes only.
_preexec_fn (Callable() | None) default=None
Private API -- provided for testing purposes only.
pty (bool | Callable(int)) default=False
If True, use a pseudo-terminal (pty) to control the child process.
If pty is set to a callable, the function must take one int argument
for the child side of the pty. The function is called to set the child
pty's termios settings before spawning the subprocess.
shellous provides three utility functions: shellous.cooked,
shellous.raw and shellous.cbreak that can be used as arguments to
the pty option.
close_fds (bool) default=True
Close all unnecessary file descriptors in the child process. This
defaults to True to align with the default behavior of the subprocess
module.
audit_callback (Callable(phase, info) | None) default=None
Specify a function to call as the command execution goes through its
lifecycle. audit_callback is a function called with two arguments,
phase and info.
phase can be one of three values:
"start": The process is about to start.
"stop": The process finally stopped.
"signal": The process is being sent a signal.
info is a dictionary providing more information for the callback. The following keys are currently defined:
"runner" (Runner): Reference to the Runner object.
"failure" (str): When phase is "stop", optional string with the
name of the exception from launching the process.
"signal" (str): When phase is "signal", the signal name/number
sent to the process, e.g. "SIGTERM".
The primary use case for audit_callback is measuring how long each
command takes to run and exporting this information to a metrics
framework like Prometheus.
coerce_arg (Callable(arg) | None) default=None
Specify a function to call on each command line argument. This function
can specify how to coerce unsupported argument types (e.g. dict) to
a sequence of strings. This function should return the original value
unchanged if there is no conversion needed.
error_limit (int | None) default=1024
Specify the number of bytes to store when redirecting STDERR to BUFFER.
After reading up to error_limit bytes, shellous will continue to
read from stderr, but will not store any additional bytes. Setting
error_limit only affects the internal BUFFER; it has no effect when
using other redirection types.
read_buffer_limit (int | None) default=65536
Specify the maximum number of bytes to read from a stdout/stderr stream
when looking for a separator. Increase this value if you expect the
subprocess to emit extremely long lines. If this value is too small,
you may get the error: "Separator is not found".
def
coro( self, *, _run_future: _asyncio.Future[Runner] | None = None) -> Coroutine[Any, Any, ~_RT]:
681 def coro( 682 self, 683 *, 684 _run_future: asyncio.Future[Runner] | None = None, 685 ) -> Coroutine[Any, Any, _RT]: 686 "Return coroutine object to run awaitable." 687 return cast( 688 Coroutine[Any, Any, _RT], 689 Runner.run_command(self, _run_future=_run_future), 690 )
Return coroutine object to run awaitable.
@contextlib.asynccontextmanager
async def
prompt( self, prompt: str | list[str] | re.Pattern[str] | None = None, *, timeout: float | None = None, normalize_newlines: bool = False) -> AsyncGenerator[Prompt, NoneType]:
692 @contextlib.asynccontextmanager 693 async def prompt( 694 self, 695 prompt: str | list[str] | re.Pattern[str] | None = None, 696 *, 697 timeout: float | None = None, 698 normalize_newlines: bool = False, 699 ) -> AsyncGenerator[Prompt, None]: 700 """Run command using the send/expect API. 701 702 This method should be called using `async with`. It returns a `Prompt` 703 object with send() and expect() methods. 704 705 You can optionally set a default `prompt`. This is used by `expect()` 706 when you don't provide another value. 707 708 Use the `timeout` parameter to set the default timeout for operations. 709 710 Set `normalize_newlines` to True to convert incoming CR and CR-LF to LF. 711 This conversion is done before matching with `expect()`. This option 712 does not affect strings sent with `send()`. 713 """ 714 cmd = self.stdin(Redirect.CAPTURE).stdout(Redirect.CAPTURE) 715 716 cli = None 717 try: 718 async with Runner(cmd) as run: 719 cli = Prompt( 720 run, 721 default_prompt=prompt, 722 default_timeout=timeout, 723 normalize_newlines=normalize_newlines, 724 ) 725 yield cli 726 cli.close() 727 finally: 728 if cli is not None: 729 cli._finish_() # pyright: ignore[reportPrivateUsage]
Run command using the send/expect API.
This method should be called using async with. It returns a Prompt
object with send() and expect() methods.
You can optionally set a default prompt. This is used by expect()
when you don't provide another value.
Use the timeout parameter to set the default timeout for operations.
Set normalize_newlines to True to convert incoming CR and CR-LF to LF.
This conversion is done before matching with expect(). This option
does not affect strings sent with send().
def
__await__(self) -> Generator[Any, NoneType, ~_RT]:
731 def __await__(self) -> "Generator[Any, None, _RT]": 732 "Run process and return the standard output." 733 return self.coro().__await__()
Run process and return the standard output.
735 async def __aenter__(self) -> Runner: 736 "Enter the async context manager." 737 return await context_aenter(self, Runner(self))
Enter the async context manager.
def
__aiter__(self) -> AsyncIterator[str]:
748 def __aiter__(self) -> AsyncIterator[str]: 749 "Return async iterator to iterate over output lines." 750 return aiter_preflight(self)._readlines()
Return async iterator to iterate over output lines.
writable: Command[~_RT]
807 @property 808 def writable(self) -> "Command[_RT]": 809 "Set `writable` to True." 810 return self.set(_writable=True)
Set writable to True.
@dataclass(frozen=True)
class
Options:
95@dataclass(frozen=True) 96class Options: # pylint: disable=too-many-instance-attributes 97 "Concrete class for per-command options." 98 99 path: str | None = None 100 "Optional search path to use instead of PATH environment variable." 101 102 env: EnvironmentDict | None = field(default=None, repr=False) 103 "Additional environment variables for command." 104 105 inherit_env: bool = True 106 "True if subprocess should inherit the current environment variables." 107 108 input: _RedirectT = Redirect.DEFAULT 109 "Input object to bind to stdin." 110 111 input_close: bool = False 112 "True if input object should be closed after subprocess launch." 113 114 output: _RedirectT = Redirect.DEFAULT 115 "Output object to bind to stdout." 116 117 output_append: bool = False 118 "True if output object should be opened in append mode." 119 120 output_close: bool = False 121 "True if output object should be closed after subprocess launch." 122 123 error: _RedirectT = Redirect.DEFAULT 124 "Error object to bind to stderr." 125 126 error_append: bool = False 127 "True if error object should be opened in append mode." 128 129 error_close: bool = False 130 "True if error object should be closed after subprocess launch." 131 132 error_limit: int | None = DEFAULT_ERROR_LIMIT 133 "Bytes of stderr to buffer in memory (`sh.BUFFER`). None means unlimited." 134 135 encoding: str = "utf-8" 136 "Specifies encoding of input/output." 137 138 _return_result: bool = False 139 "True if we should return `Result` object instead of the output text/bytes." 140 141 _catch_cancelled_error: bool = False 142 "True if we should raise `ResultError` after clean up from cancelled task." 143 144 exit_codes: Container[int] | None = None 145 "Set of exit codes that do not raise a `ResultError`. None means {0}." 146 147 timeout: float | None = None 148 "Timeout in seconds that we wait before cancelling the process." 149 150 cancel_timeout: float = 3.0 151 "Timeout in seconds that we wait for a cancelled process to terminate." 152 153 cancel_signal: signal.Signals | None = signal.SIGTERM 154 "The signal sent to terminate a cancelled process." 155 156 alt_name: str | None = None 157 "Alternate name for the command to use when logging." 158 159 pass_fds: Iterable[int] = () 160 "File descriptors to pass to the command." 161 162 pass_fds_close: bool = False 163 "True if pass_fds should be closed after subprocess launch." 164 165 _writable: bool = False 166 "True if using process substitution in write mode." 167 168 _start_new_session: bool = False 169 "True if child process should start a new session with `setsid` call." 170 171 _preexec_fn: _PreexecFnT = None 172 "Function to call in child process after fork from parent." 173 174 pty: PtyAdapterOrBool = False 175 "True if child process should be controlled using a pseudo-terminal (pty)." 176 177 close_fds: bool = True 178 "True if child process should close all file descriptors." 179 180 audit_callback: _AuditFnT = None 181 "Function called to audit stages of process execution." 182 183 coerce_arg: _CoerceArgFnT = None 184 "Function called to coerce top level arguments." 185 186 read_buffer_limit: int | None = None 187 "Maximum number of bytes to read when looking for a separator." 188 189 def runtime_env(self) -> dict[str, str] | None: 190 "@private Return our `env` merged with the global environment." 191 if self.inherit_env: 192 if not self.env: 193 return None 194 return os.environ | self.env 195 196 if self.env: 197 return dict(self.env) # make copy of dict 198 return {} 199 200 def set_stdin(self, input_: Any, close: bool) -> "Options": 201 "@private Return new options with `input` configured." 202 if input_ is None: 203 raise TypeError("invalid stdin") 204 205 if input_ == Redirect.STDOUT: 206 raise ValueError("STDOUT is only supported by stderr") 207 208 return dataclasses.replace( 209 self, 210 input=input_, 211 input_close=close, 212 ) 213 214 def set_stdout(self, output: Any, append: bool, close: bool) -> "Options": 215 "@private Return new options with `output` configured." 216 if output is None: 217 raise TypeError("invalid stdout") 218 219 if output == Redirect.STDOUT: 220 raise ValueError("STDOUT is only supported by stderr") 221 222 return dataclasses.replace( 223 self, 224 output=output, 225 output_append=append, 226 output_close=close, 227 ) 228 229 def set_stderr(self, error: Any, append: bool, close: bool) -> "Options": 230 "@private Return new options with `error` configured." 231 if error is None: 232 raise TypeError("invalid stderr") 233 234 return dataclasses.replace( 235 self, 236 error=error, 237 error_append=append, 238 error_close=close, 239 ) 240 241 def add_env(self, updates: dict[str, Any]) -> "Options": 242 "@private Return new options with augmented environment." 243 new_env = EnvironmentDict(self.env, updates) 244 return dataclasses.replace(self, env=new_env) 245 246 def set(self, kwds: dict[str, Any]) -> "Options": 247 """@private Return new options with given properties updated. 248 249 See `Command.set` for option reference. 250 """ 251 kwds = {key: value for key, value in kwds.items() if value is not _UNSET} 252 if "env" in kwds: 253 # The "env" property is stored as an `EnvironmentDict`. 254 new_env = kwds["env"] 255 if new_env: 256 kwds["env"] = EnvironmentDict(None, new_env) 257 else: 258 kwds["env"] = None 259 return dataclasses.replace(self, **kwds) 260 261 @overload 262 def which(self, name: bytes) -> bytes | None: 263 "@private Find the command with the given name and return its path." 264 265 @overload 266 def which(self, name: str | os.PathLike[Any]) -> str | os.PathLike[Any] | None: 267 "@private Find the command with the given name and return its path." 268 269 def which( 270 self, name: str | bytes | os.PathLike[Any] 271 ) -> str | bytes | os.PathLike[Any] | None: 272 "@private Find the command with the given name and return its path." 273 if sys.platform == "win32" and sys.version_info < (3, 12): 274 # On Windows before Python 3.12, using a Path for `name` causes an 275 # AttributeError. Forcibly coerce Path to String. 276 if isinstance(name, os.PathLike): 277 name = str(name) 278 return shutil.which(name, path=self.path)
Concrete class for per-command options.
Options( path: str | None = None, env: shellous.util.EnvironmentDict | None = None, inherit_env: bool = True, input: Any = <Redirect.DEFAULT: -20>, input_close: bool = False, output: Any = <Redirect.DEFAULT: -20>, output_append: bool = False, output_close: bool = False, error: Any = <Redirect.DEFAULT: -20>, error_append: bool = False, error_close: bool = False, error_limit: int | None = 1024, encoding: str = 'utf-8', _return_result: bool = False, _catch_cancelled_error: bool = False, exit_codes: Optional[Container[int]] = None, timeout: float | None = None, cancel_timeout: float = 3.0, cancel_signal: signal.Signals | None = <Signals.SIGTERM: 15>, alt_name: str | None = None, pass_fds: Iterable[int] = (), pass_fds_close: bool = False, _writable: bool = False, _start_new_session: bool = False, _preexec_fn: Optional[Callable[[], NoneType]] = None, pty: Union[Callable[[int], NoneType], bool] = False, close_fds: bool = True, audit_callback: Optional[Callable[[str, AuditEventInfo], NoneType]] = None, coerce_arg: Optional[Callable[[Any], Any]] = None, read_buffer_limit: int | None = None)
error_limit: int | None =
1024
Bytes of stderr to buffer in memory (sh.BUFFER). None means unlimited.
exit_codes: Optional[Container[int]] =
None
Set of exit codes that do not raise a ResultError. None means {0}.
cancel_signal: signal.Signals | None =
<Signals.SIGTERM: 15>
The signal sent to terminate a cancelled process.
pty: Union[Callable[[int], NoneType], bool] =
False
True if child process should be controlled using a pseudo-terminal (pty).
@dataclass(frozen=True)
class
Pipeline39@dataclass(frozen=True) 40class Pipeline(Generic[_RT]): 41 "A Pipeline is a sequence of commands." 42 43 commands: tuple[shellous.Command[_RT], ...] = () 44 45 @staticmethod 46 def create(*commands: shellous.Command[_T]) -> "Pipeline[_T]": 47 "Create a new Pipeline." 48 return Pipeline(commands) 49 50 def __post_init__(self) -> None: 51 "Validate the pipeline." 52 if len(self.commands) == 0: 53 raise ValueError("Pipeline must include at least one command") 54 55 @property 56 def name(self) -> str: 57 "Return the name of the pipeline." 58 return "|".join(cmd.name for cmd in self.commands) 59 60 @property 61 def options(self) -> shellous.Options: 62 "Return the last command's options." 63 return self.commands[-1].options 64 65 def stdin(self, input_: Any, *, close: bool = False) -> "Pipeline[_RT]": 66 "Set stdin on the first command of the pipeline." 67 new_first = self.commands[0].stdin(input_, close=close) 68 new_commands = (new_first, *self.commands[1:]) 69 return dataclasses.replace(self, commands=new_commands) 70 71 def stdout( 72 self, 73 output: Any, 74 *, 75 append: bool = False, 76 close: bool = False, 77 ) -> "Pipeline[_RT]": 78 "Set stdout on the last command of the pipeline." 79 new_last = self.commands[-1].stdout(output, append=append, close=close) 80 new_commands = (*self.commands[0:-1], new_last) 81 return dataclasses.replace(self, commands=new_commands) 82 83 def stderr( 84 self, 85 error: Any, 86 *, 87 append: bool = False, 88 close: bool = False, 89 ) -> "Pipeline[_RT]": 90 "Set stderr on the last command of the pipeline." 91 new_last = self.commands[-1].stderr(error, append=append, close=close) 92 new_commands = (*self.commands[0:-1], new_last) 93 return dataclasses.replace(self, commands=new_commands) 94 95 def _set(self, **kwds: Any): 96 "Set options on last command of the pipeline." 97 new_last = self.commands[-1].set(**kwds) 98 new_commands = (*self.commands[0:-1], new_last) 99 return dataclasses.replace(self, commands=new_commands) 100 101 def coro(self) -> Coroutine[Any, Any, _RT]: 102 "Return coroutine object for pipeline." 103 return cast(Coroutine[Any, Any, _RT], PipeRunner.run_pipeline(self)) 104 105 @contextlib.asynccontextmanager 106 async def prompt( 107 self, 108 prompt: str | list[str] | re.Pattern[str] | None = None, 109 *, 110 timeout: float | None = None, 111 normalize_newlines: bool = False, 112 ) -> AsyncGenerator[Prompt, None]: 113 """Run pipeline using the send/expect API. 114 115 This method should be called using `async with`. It returns a `Prompt` 116 object with send() and expect() methods. 117 118 You can optionally set a default `prompt`. This is used by `expect()` 119 when you don't provide another value. 120 121 Use the `timeout` parameter to set the default timeout for operations. 122 123 Set `normalize_newlines` to True to convert incoming CR and CR-LF to LF. 124 This conversion is done before matching with `expect()`. This option 125 does not affect strings sent with `send()`. 126 """ 127 cmd = self.stdin(Redirect.CAPTURE).stdout(Redirect.CAPTURE) 128 129 cli = None 130 try: 131 async with PipeRunner(cmd, capturing=True) as run: 132 cli = Prompt( 133 run, 134 default_prompt=prompt, 135 default_timeout=timeout, 136 normalize_newlines=normalize_newlines, 137 ) 138 yield cli 139 cli.close() 140 finally: 141 if cli is not None: 142 cli._finish_() # pyright: ignore[reportPrivateUsage] 143 144 def _add(self, item: "shellous.Command[Any] | Pipeline[Any]"): 145 if isinstance(item, shellous.Command): 146 return dataclasses.replace(self, commands=(*self.commands, item)) 147 return dataclasses.replace( 148 self, 149 commands=self.commands + item.commands, 150 ) 151 152 def __len__(self) -> int: 153 "Return number of commands in pipe." 154 return len(self.commands) 155 156 def __getitem__(self, key: int) -> shellous.Command[Any]: 157 "Return specified command by index." 158 return self.commands[key] 159 160 def __call__(self, *args: Any) -> "Pipeline[_RT]": 161 if args: 162 raise TypeError("Calling pipeline with 1 or more arguments.") 163 return self 164 165 @overload 166 def __or__( 167 self, rhs: "shellous.Command[shellous.Result] | Pipeline[shellous.Result]" 168 ) -> "Pipeline[shellous.Result]": ... # pragma: no cover 169 170 @overload 171 def __or__( 172 self, rhs: "shellous.Command[str] | Pipeline[str]" 173 ) -> "Pipeline[str]": ... # pragma: no cover 174 175 @overload 176 def __or__(self, rhs: StdoutType) -> "Pipeline[_RT]": ... # pragma: no cover 177 178 def __or__(self, rhs: Any) -> "Pipeline[Any]": 179 if isinstance(rhs, (shellous.Command, Pipeline)): 180 return self._add(rhs) # pyright: ignore[reportUnknownArgumentType] 181 if isinstance(rhs, STDOUT_TYPES): 182 return self.stdout(rhs) 183 if isinstance(rhs, (str, bytes)): 184 raise TypeError( 185 f"{type(rhs)!r} unsupported for | output (Use 'pathlib.Path')" 186 ) 187 return NotImplemented 188 189 def __ror__(self, lhs: StdinType) -> "Pipeline[_RT]": 190 if isinstance(lhs, STDIN_TYPES): # pyright: ignore[reportUnnecessaryIsInstance] 191 return self.stdin(lhs) 192 return NotImplemented 193 194 def __rshift__(self, rhs: StdoutType) -> "Pipeline[_RT]": 195 if isinstance( 196 rhs, STDOUT_TYPES 197 ): # pyright: ignore[reportUnnecessaryIsInstance] 198 return self.stdout(rhs, append=True) 199 if isinstance(rhs, (str, bytes)): 200 raise TypeError( 201 f"{type(rhs)!r} unsupported for >> output (Use 'pathlib.Path')" 202 ) 203 return NotImplemented 204 205 @property 206 def writable(self) -> "Pipeline[_RT]": 207 "Set writable=True option on last command of pipeline." 208 return self._set(_writable=True) 209 210 @property 211 def result(self) -> "Pipeline[shellous.Result]": 212 "Set `_return_result` and `exit_codes`." 213 return cast( 214 Pipeline[shellous.Result], 215 self._set(_return_result=True, exit_codes=range(-255, 2**32)), 216 ) 217 218 def __await__(self) -> "Generator[Any, None, _RT]": 219 return self.coro().__await__() # FP pylint: disable=no-member 220 221 async def __aenter__(self) -> PipeRunner: 222 "Enter the async context manager." 223 return await context_aenter(self, PipeRunner(self, capturing=True)) 224 225 async def __aexit__( 226 self, 227 exc_type: type[BaseException] | None, 228 exc_value: BaseException | None, 229 exc_tb: TracebackType | None, 230 ) -> bool | None: 231 "Exit the async context manager." 232 return await context_aexit(self, exc_type, exc_value, exc_tb) 233 234 def __aiter__(self) -> AsyncIterator[str]: 235 "Return async iterator to iterate over output lines." 236 return aiter_preflight(self)._readlines() 237 238 async def _readlines(self): 239 "Async generator to iterate over lines." 240 async with PipeRunner(self, capturing=True) as run: 241 async for line in run: 242 yield line
A Pipeline is a sequence of commands.
Pipeline(commands: tuple[Command[~_RT], ...] = ())
45 @staticmethod 46 def create(*commands: shellous.Command[_T]) -> "Pipeline[_T]": 47 "Create a new Pipeline." 48 return Pipeline(commands)
Create a new Pipeline.
name: str
55 @property 56 def name(self) -> str: 57 "Return the name of the pipeline." 58 return "|".join(cmd.name for cmd in self.commands)
Return the name of the pipeline.
options: Options
60 @property 61 def options(self) -> shellous.Options: 62 "Return the last command's options." 63 return self.commands[-1].options
Return the last command's options.
65 def stdin(self, input_: Any, *, close: bool = False) -> "Pipeline[_RT]": 66 "Set stdin on the first command of the pipeline." 67 new_first = self.commands[0].stdin(input_, close=close) 68 new_commands = (new_first, *self.commands[1:]) 69 return dataclasses.replace(self, commands=new_commands)
Set stdin on the first command of the pipeline.
71 def stdout( 72 self, 73 output: Any, 74 *, 75 append: bool = False, 76 close: bool = False, 77 ) -> "Pipeline[_RT]": 78 "Set stdout on the last command of the pipeline." 79 new_last = self.commands[-1].stdout(output, append=append, close=close) 80 new_commands = (*self.commands[0:-1], new_last) 81 return dataclasses.replace(self, commands=new_commands)
Set stdout on the last command of the pipeline.
83 def stderr( 84 self, 85 error: Any, 86 *, 87 append: bool = False, 88 close: bool = False, 89 ) -> "Pipeline[_RT]": 90 "Set stderr on the last command of the pipeline." 91 new_last = self.commands[-1].stderr(error, append=append, close=close) 92 new_commands = (*self.commands[0:-1], new_last) 93 return dataclasses.replace(self, commands=new_commands)
Set stderr on the last command of the pipeline.
def
coro(self) -> Coroutine[Any, Any, ~_RT]:
101 def coro(self) -> Coroutine[Any, Any, _RT]: 102 "Return coroutine object for pipeline." 103 return cast(Coroutine[Any, Any, _RT], PipeRunner.run_pipeline(self))
Return coroutine object for pipeline.
@contextlib.asynccontextmanager
async def
prompt( self, prompt: str | list[str] | re.Pattern[str] | None = None, *, timeout: float | None = None, normalize_newlines: bool = False) -> AsyncGenerator[Prompt, NoneType]:
105 @contextlib.asynccontextmanager 106 async def prompt( 107 self, 108 prompt: str | list[str] | re.Pattern[str] | None = None, 109 *, 110 timeout: float | None = None, 111 normalize_newlines: bool = False, 112 ) -> AsyncGenerator[Prompt, None]: 113 """Run pipeline using the send/expect API. 114 115 This method should be called using `async with`. It returns a `Prompt` 116 object with send() and expect() methods. 117 118 You can optionally set a default `prompt`. This is used by `expect()` 119 when you don't provide another value. 120 121 Use the `timeout` parameter to set the default timeout for operations. 122 123 Set `normalize_newlines` to True to convert incoming CR and CR-LF to LF. 124 This conversion is done before matching with `expect()`. This option 125 does not affect strings sent with `send()`. 126 """ 127 cmd = self.stdin(Redirect.CAPTURE).stdout(Redirect.CAPTURE) 128 129 cli = None 130 try: 131 async with PipeRunner(cmd, capturing=True) as run: 132 cli = Prompt( 133 run, 134 default_prompt=prompt, 135 default_timeout=timeout, 136 normalize_newlines=normalize_newlines, 137 ) 138 yield cli 139 cli.close() 140 finally: 141 if cli is not None: 142 cli._finish_() # pyright: ignore[reportPrivateUsage]
Run pipeline using the send/expect API.
This method should be called using async with. It returns a Prompt
object with send() and expect() methods.
You can optionally set a default prompt. This is used by expect()
when you don't provide another value.
Use the timeout parameter to set the default timeout for operations.
Set normalize_newlines to True to convert incoming CR and CR-LF to LF.
This conversion is done before matching with expect(). This option
does not affect strings sent with send().
def
__len__(self) -> int:
152 def __len__(self) -> int: 153 "Return number of commands in pipe." 154 return len(self.commands)
Return number of commands in pipe.
156 def __getitem__(self, key: int) -> shellous.Command[Any]: 157 "Return specified command by index." 158 return self.commands[key]
Return specified command by index.
writable: Pipeline[~_RT]
205 @property 206 def writable(self) -> "Pipeline[_RT]": 207 "Set writable=True option on last command of pipeline." 208 return self._set(_writable=True)
Set writable=True option on last command of pipeline.
210 @property 211 def result(self) -> "Pipeline[shellous.Result]": 212 "Set `_return_result` and `exit_codes`." 213 return cast( 214 Pipeline[shellous.Result], 215 self._set(_return_result=True, exit_codes=range(-255, 2**32)), 216 )
Set _return_result and exit_codes.
class
Prompt:
23class Prompt: 24 """Utility class to help with an interactive prompt session. 25 26 When you are controlling a co-process you will usually "send" some 27 text to it, and then "expect" a response. The "expect" operation can use 28 a regular expression to match different types of responses. 29 30 Create a new `Prompt` instance using the `prompt()` API. 31 32 ``` 33 # In this example, we are using a default prompt of "??? ". 34 # Setting the PS1 environment variable tells the shell to use this as 35 # the shell prompt. 36 cmd = sh("sh").env(PS1="??? ").set(pty=True) 37 38 async with cmd.prompt("??? ", timeout=3.0) as cli: 39 # Turn off terminal echo. 40 cli.echo = False 41 42 # Wait for greeting and initial prompt. Calling expect() with no 43 # argument will match the default prompt "??? ". 44 greeting, _ = await cli.expect() 45 46 # Send a command and wait for the response. 47 await cli.send("echo hello") 48 answer, _ = await cli.expect() 49 assert answer == "hello\\r\\n" 50 ``` 51 """ 52 53 _runner: Runner | PipeRunner 54 _encoding: str 55 _default_prompt: re.Pattern[str] | None 56 _default_timeout: float | None 57 _normalize_newlines: bool 58 _chunk_size: int 59 _decoder: codecs.IncrementalDecoder 60 _pending: str = "" 61 _at_eof: bool = False 62 _result: Result | None = None 63 64 def __init__( 65 self, 66 runner: Runner, 67 *, 68 default_prompt: str | list[str] | re.Pattern[str] | None = None, 69 default_timeout: float | None = None, 70 normalize_newlines: bool = False, 71 _chunk_size: int = _DEFAULT_CHUNK_SIZE, 72 ): 73 assert runner.stdin is not None 74 assert runner.stdout is not None 75 76 if isinstance(default_prompt, (str, list)): 77 default_prompt = _regex_compile_exact(default_prompt) 78 79 self._runner = runner 80 self._encoding = runner.options.encoding 81 self._default_prompt = default_prompt 82 self._default_timeout = default_timeout 83 self._normalize_newlines = normalize_newlines 84 self._chunk_size = _chunk_size 85 self._decoder = _make_decoder(self._encoding, normalize_newlines) 86 87 if LOG_PROMPT: 88 LOGGER.info( 89 "Prompt[pid=%s]: --- BEGIN --- name=%r", 90 self._runner.pid, 91 self._runner.name, 92 ) 93 94 @property 95 def at_eof(self) -> bool: 96 "True if the prompt reader is at the end of file." 97 return self._at_eof and not self._pending 98 99 @property 100 def pending(self) -> str: 101 """Characters that still remain in the `pending` buffer.""" 102 return self._pending 103 104 @property 105 def echo(self) -> bool: 106 """True if PTY is in echo mode. 107 108 If the runner is not using a PTY, always return False. 109 110 When the process is using a PTY, you can enable/disable terminal echo 111 mode by setting the `echo` property to True/False. 112 """ 113 if not isinstance(self._runner, Runner) or self._runner.pty_fd is None: 114 return False 115 return pty_util.get_term_echo(self._runner.pty_fd) 116 117 @echo.setter 118 def echo(self, value: bool) -> None: 119 """Set echo mode for the PTY. 120 121 Raise an error if the runner is not using a PTY. 122 """ 123 if not isinstance(self._runner, Runner) or self._runner.pty_fd is None: 124 raise RuntimeError("Cannot set echo mode. Not running in a PTY.") 125 pty_util.set_term_echo(self._runner.pty_fd, value) 126 127 @property 128 def result(self) -> Result: 129 """The `Result` of the co-process when it exited. 130 131 You can only retrieve this property *after* the `async with` block 132 exits where the co-process is running: 133 134 ``` 135 async with cmd.prompt() as cli: 136 ... 137 # Access `cli.result` here. 138 ``` 139 140 Inside the `async with` block, raise a RuntimeError because the process 141 has not exited yet. 142 """ 143 if self._result is None: 144 raise RuntimeError("Prompt process is still running") 145 return self._result 146 147 async def send( 148 self, 149 text: bytes | str, 150 *, 151 end: str | None = _DEFAULT_LINE_END, 152 no_echo: bool = False, 153 timeout: float | None = None, 154 ) -> None: 155 """Write some `text` to co-process standard input and append a newline. 156 157 The `text` parameter is the string that you want to write. Use a `bytes` 158 object to send some raw bytes (e.g. to the terminal driver). 159 160 The default line ending is "\n". Use the `end` parameter to change the 161 line ending. To omit the line ending entirely, specify `end=None`. 162 163 Set `no_echo` to True when you are writing a password. When you set 164 `no_echo` to True, the `send` method will wait for terminal 165 echo mode to be disabled before writing the text. If shellous logging 166 is enabled, the sensitive information will **not** be logged. 167 168 Use the `timeout` parameter to override the default timeout. Normally, 169 data is delivered immediately and this method returns making a trip 170 through the event loop. However, there are situations where the 171 co-process input pipeline fills and we have to wait for it to 172 drain. 173 174 When this method needs to wait for the co-process input pipe to drain, 175 this method will concurrently read from the output pipe into the pending 176 buffer. This is necessary to avoid a deadlock situation where everything 177 stops because neither process can make progress. 178 """ 179 if end is None: 180 end = "" 181 182 if no_echo: 183 await self._wait_no_echo() 184 185 if isinstance(text, bytes): 186 data = text + encode_bytes(end, self._encoding) 187 else: 188 data = encode_bytes(text + end, self._encoding) 189 190 stdin = self._runner.stdin 191 assert stdin is not None 192 stdin.write(data) 193 194 if LOG_PROMPT: 195 self._log_send(data, no_echo) 196 197 # Drain our write to stdin. 198 cancelled, ex = await harvest_results( 199 self._drain(stdin), 200 timeout=timeout or self._default_timeout, 201 ) 202 if cancelled: 203 raise asyncio.CancelledError() 204 if isinstance(ex[0], Exception): 205 raise ex[0] 206 207 async def expect( 208 self, 209 prompt: str | list[str] | re.Pattern[str] | None = None, 210 *, 211 timeout: float | None = None, 212 ) -> tuple[str, re.Match[str]]: 213 """Read from co-process standard output until `prompt` pattern matches. 214 215 Returns a 2-tuple of (output, match) where `output` is the text *before* 216 the prompt pattern and `match` is a `re.Match` object for the prompt 217 text itself. 218 219 If `expect` reaches EOF or a timeout occurs before the prompt pattern 220 matches, it raises an `EOFError` or `asyncio.TimeoutError`. The unread 221 characters will be available in the `pending` buffer. 222 223 After this method returns, there may still be characters read from stdout 224 that remain in the `pending` buffer. These are the characters *after* the 225 prompt pattern. You can examine these using the `pending` property. 226 Subsequent calls to expect will examine this buffer first before 227 reading new data from the output pipe. 228 229 By default, this method will use the default `timeout` for the `Prompt` 230 object if one is set. You can use the `timeout` parameter to specify 231 a custom timeout in seconds. 232 233 Prompt Patterns 234 ~~~~~~~~~~~~~~~ 235 236 The `expect()` method supports matching fixed strings and regular 237 expressions. The type of the `prompt` parameter determines the type of 238 search. 239 240 No argument or `None`: 241 Use the default prompt pattern. If there is no default prompt, 242 raise a TypeError. 243 `str`: 244 Match this string exactly. 245 `list[str]`: 246 Match one of these strings exactly. 247 `re.Pattern[str]`: 248 Match the given regular expression. 249 250 When matching a regular expression, only a single Pattern object is 251 supported. To match multiple regular expressions, combine them into a 252 single regular expression using *alternation* syntax (|). 253 254 The `expect()` method returns a 2-tuple (output, match). The `match` 255 is the result of the regular expression search (re.Match). If you 256 specify your prompt as a string or list of strings, it is still compiled 257 into a regular expression that produces an `re.Match` object. You can 258 examine the `match` object to determine the prompt value found. 259 260 This method conducts a regular expression search on streaming data. The 261 `expect()` method reads a new chunk of data into the `pending` buffer 262 and then searches it. You must be careful in writing a regular 263 expression so that the search is agnostic to how the incoming chunks of 264 data arrive. Consider including a boundary condition at the end of your pattern. 265 For example, instead of searching for the open-ended pattern`[a-z]+`, 266 search for the pattern `[a-z]+[^a-z]` which ends with a non-letter 267 character. 268 269 Examples 270 ~~~~~~~~ 271 272 Expect an exact string: 273 274 ``` 275 await cli.expect("ftp> ") 276 await cli.send(command) 277 response, _ = await cli.expect("ftp> ") 278 ``` 279 280 Expect a choice of strings: 281 282 ``` 283 _, m = await cli.expect(["Login: ", "Password: ", "ftp> "]) 284 match m[0]: 285 case "Login: ": 286 await cli.send(login) 287 case "Password: ": 288 await cli.send(password) 289 case "ftp> ": 290 await cli.send(command) 291 ``` 292 293 Read until EOF: 294 295 ``` 296 data = await cli.read_all() 297 ``` 298 299 Read the contents of the `pending` buffer without filling the buffer 300 with any new data from the co-process pipe: 301 302 ``` 303 data = await cli.read_pending() 304 ``` 305 """ 306 if prompt is None: 307 prompt = self._default_prompt 308 if prompt is None: 309 raise TypeError("prompt is required when default prompt is not set") 310 elif isinstance(prompt, (str, list)): 311 prompt = _regex_compile_exact(prompt) 312 313 if self._pending: 314 result = self._search_pending(prompt) 315 if result is not None: 316 return result 317 318 if self._at_eof: 319 raise EOFError("Prompt has reached EOF") 320 321 cancelled, (result,) = await harvest_results( 322 self._read_to_pattern(prompt), 323 timeout=timeout or self._default_timeout, 324 ) 325 if cancelled: 326 raise asyncio.CancelledError() 327 if isinstance(result, Exception): 328 raise result 329 330 return result 331 332 async def read_all( 333 self, 334 *, 335 timeout: float | None = None, 336 ) -> str: 337 """Read from co-process output until EOF. 338 339 If we are already at EOF, return "". 340 """ 341 if not self._at_eof: 342 cancelled, (result,) = await harvest_results( 343 self._read_some(tag="@read_all"), 344 timeout=timeout or self._default_timeout, 345 ) 346 if cancelled: 347 raise asyncio.CancelledError() 348 if isinstance(result, Exception): 349 raise result 350 351 return self.read_pending() 352 353 def read_pending(self) -> str: 354 """Read the contents of the pending buffer and empty it. 355 356 This method does not fill the pending buffer with any new data from the 357 co-process output pipe. If the pending buffer is already empty, return 358 "". 359 """ 360 result = self._pending 361 self._pending = "" 362 return result 363 364 async def command( 365 self, 366 text: str, 367 *, 368 end: str = _DEFAULT_LINE_END, 369 no_echo: bool = False, 370 prompt: str | re.Pattern[str] | None = None, 371 timeout: float | None = None, 372 allow_eof: bool = False, 373 ) -> str: 374 """Send some text to the co-process and return the response. 375 376 This method is equivalent to calling send() following by expect(). 377 However, the return value is simpler; `command()` does not return the 378 `re.Match` object. 379 380 If you call this method *after* the co-process output pipe has already 381 returned EOF, raise `EOFError`. 382 383 If `allow_eof` is True, this method will read data up to EOF instead of 384 raising an EOFError. 385 """ 386 if self._at_eof: 387 raise EOFError("Prompt has reached EOF") 388 389 await self.send(text, end=end, no_echo=no_echo, timeout=timeout) 390 try: 391 result, _ = await self.expect(prompt, timeout=timeout) 392 except EOFError: 393 if not allow_eof: 394 raise 395 result = self.read_pending() 396 397 return result 398 399 def close(self) -> None: 400 "Close stdin to end the prompt session." 401 stdin = self._runner.stdin 402 assert stdin is not None 403 404 if isinstance(self._runner, Runner) and self._runner.pty_eof: 405 # Write EOF twice; once to end the current line, and the second 406 # time to signal the end. 407 stdin.write(self._runner.pty_eof * 2) 408 if LOG_PROMPT: 409 LOGGER.info("Prompt[pid=%s] send: [[EOF]]", self._runner.pid) 410 411 else: 412 stdin.close() 413 if LOG_PROMPT: 414 LOGGER.info("Prompt[pid=%s] close", self._runner.pid) 415 416 def _finish_(self) -> None: 417 "Internal method called when process exits to fetch the `Result` and cache it." 418 self._result = self._runner.result(check=False) 419 if LOG_PROMPT: 420 LOGGER.info( 421 "Prompt[pid=%s]: --- END --- result=%r", 422 self._runner.pid, 423 self._result, 424 ) 425 426 async def _read_to_pattern( 427 self, 428 pattern: re.Pattern[str], 429 ) -> tuple[str, re.Match[str]]: 430 """Read text up to part that matches the pattern. 431 432 Returns 2-tuple with (text, match). 433 """ 434 stdout = self._runner.stdout 435 assert stdout is not None 436 assert self._chunk_size > 0 437 438 while not self._at_eof: 439 _prev_len = len(self._pending) # debug check 440 441 try: 442 # Read chunk and check for EOF. 443 chunk = await stdout.read(self._chunk_size) 444 if not chunk: 445 self._at_eof = True 446 except asyncio.CancelledError: 447 if LOG_PROMPT: 448 LOGGER.info( 449 "Prompt[pid=%s] receive cancelled: pending=%r", 450 self._runner.pid, 451 self._pending, 452 ) 453 raise 454 455 if LOG_PROMPT: 456 self._log_receive(chunk) 457 458 # Decode eligible bytes into our buffer. 459 data = self._decoder.decode(chunk, final=self._at_eof) 460 if not data and not self._at_eof: 461 continue 462 self._pending += data 463 464 result = self._search_pending(pattern) 465 if result is not None: 466 return result 467 468 assert self._at_eof or len(self._pending) > _prev_len # debug check 469 470 raise EOFError("Prompt has reached EOF") 471 472 def _search_pending( 473 self, 474 pattern: re.Pattern[str], 475 ) -> tuple[str, re.Match[str]] | None: 476 """Search our `pending` buffer for the pattern. 477 478 If we find a match, we return the data up to the portion that matched 479 and leave the trailing data in the `pending` buffer. This method returns 480 (result, match) or None if there is no match. 481 """ 482 found = pattern.search(self._pending) 483 if found: 484 result = self._pending[0 : found.start(0)] 485 self._pending = self._pending[found.end(0) :] 486 if LOG_PROMPT: 487 LOGGER.info( 488 "Prompt[pid=%s] found: %r [%s CHARS PENDING]", 489 self._runner.pid, 490 found, 491 len(self._pending), 492 ) 493 return (result, found) 494 495 return None 496 497 async def _drain(self, stream: asyncio.StreamWriter) -> None: 498 "Drain stream while reading into buffer concurrently." 499 read_task = asyncio.create_task( 500 self._read_some(tag="@drain", concurrent_cancel=True) 501 ) 502 try: 503 await stream.drain() 504 505 finally: 506 if not read_task.done(): 507 read_task.cancel() 508 await read_task 509 510 async def _read_some( 511 self, 512 *, 513 tag: str = "", 514 concurrent_cancel: bool = False, 515 ) -> None: 516 "Read into `pending` buffer until cancelled or EOF." 517 stdout = self._runner.stdout 518 assert stdout is not None 519 assert self._chunk_size > 0 520 521 while not self._at_eof: 522 # Yield time to other tasks; read() doesn't yield as long as there 523 # is data to read. We need to provide a cancel point when this 524 # method is called during `drain`. 525 if concurrent_cancel: 526 await asyncio.sleep(0) 527 528 # Read chunk and check for EOF. 529 chunk = await stdout.read(self._chunk_size) 530 if not chunk: 531 self._at_eof = True 532 533 if LOG_PROMPT: 534 self._log_receive(chunk, tag) 535 536 # Decode eligible bytes into our buffer. 537 data = self._decoder.decode(chunk, final=self._at_eof) 538 self._pending += data 539 540 async def _wait_no_echo(self): 541 "Wait for terminal echo mode to be disabled." 542 if LOG_PROMPT: 543 LOGGER.info("Prompt[pid=%s] wait: no_echo", self._runner.pid) 544 545 for _ in range(4 * 30): 546 if not self.echo: 547 break 548 await asyncio.sleep(0.25) 549 else: 550 raise RuntimeError("Timed out: Terminal echo mode remains enabled.") 551 552 def _log_send(self, data: bytes, no_echo: bool): 553 "Log data as it is being sent." 554 pid = self._runner.pid 555 556 if no_echo: 557 LOGGER.info("Prompt[pid=%s] send: [[HIDDEN]]", pid) 558 else: 559 data_len = len(data) 560 if data_len > _LOG_LIMIT: 561 LOGGER.info( 562 "Prompt[pid=%s] send: [%d B] %r...%r", 563 pid, 564 data_len, 565 data[: _LOG_LIMIT - _LOG_LIMIT_END], 566 data[-_LOG_LIMIT_END:], 567 ) 568 else: 569 LOGGER.info( 570 "Prompt[pid=%s] send: [%d B] %r", 571 pid, 572 data_len, 573 data, 574 ) 575 576 def _log_receive(self, data: bytes, tag: str = ""): 577 "Log data as it is being received." 578 pid = self._runner.pid 579 data_len = len(data) 580 581 if data_len > _LOG_LIMIT: 582 LOGGER.info( 583 "Prompt[pid=%s] receive%s: [%d B] %r...%r", 584 pid, 585 tag, 586 data_len, 587 data[: _LOG_LIMIT - _LOG_LIMIT_END], 588 data[-_LOG_LIMIT_END:], 589 ) 590 else: 591 LOGGER.info( 592 "Prompt[pid=%s] receive%s: [%d B] %r", 593 pid, 594 tag, 595 data_len, 596 data, 597 )
Utility class to help with an interactive prompt session.
When you are controlling a co-process you will usually "send" some text to it, and then "expect" a response. The "expect" operation can use a regular expression to match different types of responses.
Create a new Prompt instance using the prompt() API.
# In this example, we are using a default prompt of "??? ".
# Setting the PS1 environment variable tells the shell to use this as
# the shell prompt.
cmd = sh("sh").env(PS1="??? ").set(pty=True)
async with cmd.prompt("??? ", timeout=3.0) as cli:
# Turn off terminal echo.
cli.echo = False
# Wait for greeting and initial prompt. Calling expect() with no
# argument will match the default prompt "??? ".
greeting, _ = await cli.expect()
# Send a command and wait for the response.
await cli.send("echo hello")
answer, _ = await cli.expect()
assert answer == "hello\r\n"
Prompt( runner: Runner, *, default_prompt: str | list[str] | re.Pattern[str] | None = None, default_timeout: float | None = None, normalize_newlines: bool = False, _chunk_size: int = 16384)
64 def __init__( 65 self, 66 runner: Runner, 67 *, 68 default_prompt: str | list[str] | re.Pattern[str] | None = None, 69 default_timeout: float | None = None, 70 normalize_newlines: bool = False, 71 _chunk_size: int = _DEFAULT_CHUNK_SIZE, 72 ): 73 assert runner.stdin is not None 74 assert runner.stdout is not None 75 76 if isinstance(default_prompt, (str, list)): 77 default_prompt = _regex_compile_exact(default_prompt) 78 79 self._runner = runner 80 self._encoding = runner.options.encoding 81 self._default_prompt = default_prompt 82 self._default_timeout = default_timeout 83 self._normalize_newlines = normalize_newlines 84 self._chunk_size = _chunk_size 85 self._decoder = _make_decoder(self._encoding, normalize_newlines) 86 87 if LOG_PROMPT: 88 LOGGER.info( 89 "Prompt[pid=%s]: --- BEGIN --- name=%r", 90 self._runner.pid, 91 self._runner.name, 92 )
at_eof: bool
94 @property 95 def at_eof(self) -> bool: 96 "True if the prompt reader is at the end of file." 97 return self._at_eof and not self._pending
True if the prompt reader is at the end of file.
pending: str
99 @property 100 def pending(self) -> str: 101 """Characters that still remain in the `pending` buffer.""" 102 return self._pending
Characters that still remain in the pending buffer.
echo: bool
104 @property 105 def echo(self) -> bool: 106 """True if PTY is in echo mode. 107 108 If the runner is not using a PTY, always return False. 109 110 When the process is using a PTY, you can enable/disable terminal echo 111 mode by setting the `echo` property to True/False. 112 """ 113 if not isinstance(self._runner, Runner) or self._runner.pty_fd is None: 114 return False 115 return pty_util.get_term_echo(self._runner.pty_fd)
True if PTY is in echo mode.
If the runner is not using a PTY, always return False.
When the process is using a PTY, you can enable/disable terminal echo
mode by setting the echo property to True/False.
result: Result
127 @property 128 def result(self) -> Result: 129 """The `Result` of the co-process when it exited. 130 131 You can only retrieve this property *after* the `async with` block 132 exits where the co-process is running: 133 134 ``` 135 async with cmd.prompt() as cli: 136 ... 137 # Access `cli.result` here. 138 ``` 139 140 Inside the `async with` block, raise a RuntimeError because the process 141 has not exited yet. 142 """ 143 if self._result is None: 144 raise RuntimeError("Prompt process is still running") 145 return self._result
The Result of the co-process when it exited.
You can only retrieve this property after the async with block
exits where the co-process is running:
async with cmd.prompt() as cli:
...
# Access `cli.result` here.
Inside the async with block, raise a RuntimeError because the process
has not exited yet.
async def
send( self, text: bytes | str, *, end: str | None = '\n', no_echo: bool = False, timeout: float | None = None) -> None:
147 async def send( 148 self, 149 text: bytes | str, 150 *, 151 end: str | None = _DEFAULT_LINE_END, 152 no_echo: bool = False, 153 timeout: float | None = None, 154 ) -> None: 155 """Write some `text` to co-process standard input and append a newline. 156 157 The `text` parameter is the string that you want to write. Use a `bytes` 158 object to send some raw bytes (e.g. to the terminal driver). 159 160 The default line ending is "\n". Use the `end` parameter to change the 161 line ending. To omit the line ending entirely, specify `end=None`. 162 163 Set `no_echo` to True when you are writing a password. When you set 164 `no_echo` to True, the `send` method will wait for terminal 165 echo mode to be disabled before writing the text. If shellous logging 166 is enabled, the sensitive information will **not** be logged. 167 168 Use the `timeout` parameter to override the default timeout. Normally, 169 data is delivered immediately and this method returns making a trip 170 through the event loop. However, there are situations where the 171 co-process input pipeline fills and we have to wait for it to 172 drain. 173 174 When this method needs to wait for the co-process input pipe to drain, 175 this method will concurrently read from the output pipe into the pending 176 buffer. This is necessary to avoid a deadlock situation where everything 177 stops because neither process can make progress. 178 """ 179 if end is None: 180 end = "" 181 182 if no_echo: 183 await self._wait_no_echo() 184 185 if isinstance(text, bytes): 186 data = text + encode_bytes(end, self._encoding) 187 else: 188 data = encode_bytes(text + end, self._encoding) 189 190 stdin = self._runner.stdin 191 assert stdin is not None 192 stdin.write(data) 193 194 if LOG_PROMPT: 195 self._log_send(data, no_echo) 196 197 # Drain our write to stdin. 198 cancelled, ex = await harvest_results( 199 self._drain(stdin), 200 timeout=timeout or self._default_timeout, 201 ) 202 if cancelled: 203 raise asyncio.CancelledError() 204 if isinstance(ex[0], Exception): 205 raise ex[0]
Write some text to co-process standard input and append a newline.
The `text` parameter is the string that you want to write. Use a `bytes`
object to send some raw bytes (e.g. to the terminal driver).
The default line ending is "
". Use the end parameter to change the
line ending. To omit the line ending entirely, specify end=None.
Set `no_echo` to True when you are writing a password. When you set
`no_echo` to True, the `send` method will wait for terminal
echo mode to be disabled before writing the text. If shellous logging
is enabled, the sensitive information will **not** be logged.
Use the `timeout` parameter to override the default timeout. Normally,
data is delivered immediately and this method returns making a trip
through the event loop. However, there are situations where the
co-process input pipeline fills and we have to wait for it to
drain.
When this method needs to wait for the co-process input pipe to drain,
this method will concurrently read from the output pipe into the pending
buffer. This is necessary to avoid a deadlock situation where everything
stops because neither process can make progress.
async def
expect( self, prompt: str | list[str] | re.Pattern[str] | None = None, *, timeout: float | None = None) -> tuple[str, re.Match[str]]:
207 async def expect( 208 self, 209 prompt: str | list[str] | re.Pattern[str] | None = None, 210 *, 211 timeout: float | None = None, 212 ) -> tuple[str, re.Match[str]]: 213 """Read from co-process standard output until `prompt` pattern matches. 214 215 Returns a 2-tuple of (output, match) where `output` is the text *before* 216 the prompt pattern and `match` is a `re.Match` object for the prompt 217 text itself. 218 219 If `expect` reaches EOF or a timeout occurs before the prompt pattern 220 matches, it raises an `EOFError` or `asyncio.TimeoutError`. The unread 221 characters will be available in the `pending` buffer. 222 223 After this method returns, there may still be characters read from stdout 224 that remain in the `pending` buffer. These are the characters *after* the 225 prompt pattern. You can examine these using the `pending` property. 226 Subsequent calls to expect will examine this buffer first before 227 reading new data from the output pipe. 228 229 By default, this method will use the default `timeout` for the `Prompt` 230 object if one is set. You can use the `timeout` parameter to specify 231 a custom timeout in seconds. 232 233 Prompt Patterns 234 ~~~~~~~~~~~~~~~ 235 236 The `expect()` method supports matching fixed strings and regular 237 expressions. The type of the `prompt` parameter determines the type of 238 search. 239 240 No argument or `None`: 241 Use the default prompt pattern. If there is no default prompt, 242 raise a TypeError. 243 `str`: 244 Match this string exactly. 245 `list[str]`: 246 Match one of these strings exactly. 247 `re.Pattern[str]`: 248 Match the given regular expression. 249 250 When matching a regular expression, only a single Pattern object is 251 supported. To match multiple regular expressions, combine them into a 252 single regular expression using *alternation* syntax (|). 253 254 The `expect()` method returns a 2-tuple (output, match). The `match` 255 is the result of the regular expression search (re.Match). If you 256 specify your prompt as a string or list of strings, it is still compiled 257 into a regular expression that produces an `re.Match` object. You can 258 examine the `match` object to determine the prompt value found. 259 260 This method conducts a regular expression search on streaming data. The 261 `expect()` method reads a new chunk of data into the `pending` buffer 262 and then searches it. You must be careful in writing a regular 263 expression so that the search is agnostic to how the incoming chunks of 264 data arrive. Consider including a boundary condition at the end of your pattern. 265 For example, instead of searching for the open-ended pattern`[a-z]+`, 266 search for the pattern `[a-z]+[^a-z]` which ends with a non-letter 267 character. 268 269 Examples 270 ~~~~~~~~ 271 272 Expect an exact string: 273 274 ``` 275 await cli.expect("ftp> ") 276 await cli.send(command) 277 response, _ = await cli.expect("ftp> ") 278 ``` 279 280 Expect a choice of strings: 281 282 ``` 283 _, m = await cli.expect(["Login: ", "Password: ", "ftp> "]) 284 match m[0]: 285 case "Login: ": 286 await cli.send(login) 287 case "Password: ": 288 await cli.send(password) 289 case "ftp> ": 290 await cli.send(command) 291 ``` 292 293 Read until EOF: 294 295 ``` 296 data = await cli.read_all() 297 ``` 298 299 Read the contents of the `pending` buffer without filling the buffer 300 with any new data from the co-process pipe: 301 302 ``` 303 data = await cli.read_pending() 304 ``` 305 """ 306 if prompt is None: 307 prompt = self._default_prompt 308 if prompt is None: 309 raise TypeError("prompt is required when default prompt is not set") 310 elif isinstance(prompt, (str, list)): 311 prompt = _regex_compile_exact(prompt) 312 313 if self._pending: 314 result = self._search_pending(prompt) 315 if result is not None: 316 return result 317 318 if self._at_eof: 319 raise EOFError("Prompt has reached EOF") 320 321 cancelled, (result,) = await harvest_results( 322 self._read_to_pattern(prompt), 323 timeout=timeout or self._default_timeout, 324 ) 325 if cancelled: 326 raise asyncio.CancelledError() 327 if isinstance(result, Exception): 328 raise result 329 330 return result
Read from co-process standard output until prompt pattern matches.
Returns a 2-tuple of (output, match) where output is the text before
the prompt pattern and match is a re.Match object for the prompt
text itself.
If expect reaches EOF or a timeout occurs before the prompt pattern
matches, it raises an EOFError or asyncio.TimeoutError. The unread
characters will be available in the pending buffer.
After this method returns, there may still be characters read from stdout
that remain in the pending buffer. These are the characters after the
prompt pattern. You can examine these using the pending property.
Subsequent calls to expect will examine this buffer first before
reading new data from the output pipe.
By default, this method will use the default timeout for the Prompt
object if one is set. You can use the timeout parameter to specify
a custom timeout in seconds.
Prompt Patterns
~~~
The expect() method supports matching fixed strings and regular
expressions. The type of the prompt parameter determines the type of
search.
No argument or None:
Use the default prompt pattern. If there is no default prompt,
raise a TypeError.
str:
Match this string exactly.
list[str]:
Match one of these strings exactly.
re.Pattern[str]:
Match the given regular expression.
When matching a regular expression, only a single Pattern object is supported. To match multiple regular expressions, combine them into a single regular expression using alternation syntax (|).
The expect() method returns a 2-tuple (output, match). The match
is the result of the regular expression search (re.Match). If you
specify your prompt as a string or list of strings, it is still compiled
into a regular expression that produces an re.Match object. You can
examine the match object to determine the prompt value found.
This method conducts a regular expression search on streaming data. The
expect() method reads a new chunk of data into the pending buffer
and then searches it. You must be careful in writing a regular
expression so that the search is agnostic to how the incoming chunks of
data arrive. Consider including a boundary condition at the end of your pattern.
For example, instead of searching for the open-ended pattern[a-z]+,
search for the pattern [a-z]+[^a-z] which ends with a non-letter
character.
Examples
~~~~
Expect an exact string:
await cli.expect("ftp> ")
await cli.send(command)
response, _ = await cli.expect("ftp> ")
Expect a choice of strings:
_, m = await cli.expect(["Login: ", "Password: ", "ftp> "])
match m[0]:
case "Login: ":
await cli.send(login)
case "Password: ":
await cli.send(password)
case "ftp> ":
await cli.send(command)
Read until EOF:
data = await cli.read_all()
Read the contents of the pending buffer without filling the buffer
with any new data from the co-process pipe:
data = await cli.read_pending()
async def
read_all(self, *, timeout: float | None = None) -> str:
332 async def read_all( 333 self, 334 *, 335 timeout: float | None = None, 336 ) -> str: 337 """Read from co-process output until EOF. 338 339 If we are already at EOF, return "". 340 """ 341 if not self._at_eof: 342 cancelled, (result,) = await harvest_results( 343 self._read_some(tag="@read_all"), 344 timeout=timeout or self._default_timeout, 345 ) 346 if cancelled: 347 raise asyncio.CancelledError() 348 if isinstance(result, Exception): 349 raise result 350 351 return self.read_pending()
Read from co-process output until EOF.
If we are already at EOF, return "".
def
read_pending(self) -> str:
353 def read_pending(self) -> str: 354 """Read the contents of the pending buffer and empty it. 355 356 This method does not fill the pending buffer with any new data from the 357 co-process output pipe. If the pending buffer is already empty, return 358 "". 359 """ 360 result = self._pending 361 self._pending = "" 362 return result
Read the contents of the pending buffer and empty it.
This method does not fill the pending buffer with any new data from the co-process output pipe. If the pending buffer is already empty, return "".
async def
command( self, text: str, *, end: str = '\n', no_echo: bool = False, prompt: str | re.Pattern[str] | None = None, timeout: float | None = None, allow_eof: bool = False) -> str:
364 async def command( 365 self, 366 text: str, 367 *, 368 end: str = _DEFAULT_LINE_END, 369 no_echo: bool = False, 370 prompt: str | re.Pattern[str] | None = None, 371 timeout: float | None = None, 372 allow_eof: bool = False, 373 ) -> str: 374 """Send some text to the co-process and return the response. 375 376 This method is equivalent to calling send() following by expect(). 377 However, the return value is simpler; `command()` does not return the 378 `re.Match` object. 379 380 If you call this method *after* the co-process output pipe has already 381 returned EOF, raise `EOFError`. 382 383 If `allow_eof` is True, this method will read data up to EOF instead of 384 raising an EOFError. 385 """ 386 if self._at_eof: 387 raise EOFError("Prompt has reached EOF") 388 389 await self.send(text, end=end, no_echo=no_echo, timeout=timeout) 390 try: 391 result, _ = await self.expect(prompt, timeout=timeout) 392 except EOFError: 393 if not allow_eof: 394 raise 395 result = self.read_pending() 396 397 return result
Send some text to the co-process and return the response.
This method is equivalent to calling send() following by expect().
However, the return value is simpler; command() does not return the
re.Match object.
If you call this method after the co-process output pipe has already
returned EOF, raise EOFError.
If allow_eof is True, this method will read data up to EOF instead of
raising an EOFError.
def
close(self) -> None:
399 def close(self) -> None: 400 "Close stdin to end the prompt session." 401 stdin = self._runner.stdin 402 assert stdin is not None 403 404 if isinstance(self._runner, Runner) and self._runner.pty_eof: 405 # Write EOF twice; once to end the current line, and the second 406 # time to signal the end. 407 stdin.write(self._runner.pty_eof * 2) 408 if LOG_PROMPT: 409 LOGGER.info("Prompt[pid=%s] send: [[EOF]]", self._runner.pid) 410 411 else: 412 stdin.close() 413 if LOG_PROMPT: 414 LOGGER.info("Prompt[pid=%s] close", self._runner.pid)
Close stdin to end the prompt session.
def
cbreak(rows: int = 0, cols: int = 0) -> Callable[[int], NoneType]:
200def cbreak(rows: int = 0, cols: int = 0) -> PtyAdapter: 201 "Return a function that sets PtyOptions.child_fd to cbreak mode." 202 203 def _pty_set_cbreak(fdesc: int): 204 tty.setcbreak(fdesc) 205 if rows or cols: 206 _set_term_size(fdesc, rows, cols) 207 assert _get_eof(fdesc) == b"" 208 209 return _pty_set_cbreak
Return a function that sets PtyOptions.child_fd to cbreak mode.
def
cooked( rows: int = 0, cols: int = 0, echo: bool = True) -> Callable[[int], NoneType]:
212def cooked(rows: int = 0, cols: int = 0, echo: bool = True) -> PtyAdapter: 213 "Return a function that leaves PtyOptions.child_fd in cooked mode." 214 215 def _pty_set_canonical(fdesc: int): 216 if rows or cols: 217 _set_term_size(fdesc, rows, cols) 218 if not echo: 219 set_term_echo(fdesc, False) 220 assert _get_eof(fdesc) == b"\x04" 221 222 return _pty_set_canonical
Return a function that leaves PtyOptions.child_fd in cooked mode.
def
raw(rows: int = 0, cols: int = 0) -> Callable[[int], NoneType]:
188def raw(rows: int = 0, cols: int = 0) -> PtyAdapter: 189 "Return a function that sets PtyOptions.child_fd to raw mode." 190 191 def _pty_set_raw(fdesc: int): 192 tty.setraw(fdesc) 193 if rows or cols: 194 _set_term_size(fdesc, rows, cols) 195 assert _get_eof(fdesc) == b"" 196 197 return _pty_set_raw
Return a function that sets PtyOptions.child_fd to raw mode.
@dataclass(frozen=True, **_KW_ONLY)
class
Result:
25@dataclass(frozen=True, **_KW_ONLY) 26class Result: 27 "Concrete class for the result of a Command." 28 29 exit_code: int 30 "Command's exit status. If < 0, this is a negated signal number." 31 32 output_bytes: bytes 33 "Output of command as bytes. May be None if there is no output." 34 35 error_bytes: bytes 36 "Limited standard error from command if not redirected." 37 38 cancelled: bool 39 "Command was cancelled." 40 41 encoding: str 42 "Output encoding." 43 44 @property 45 def output(self) -> str: 46 "Output of command as a string." 47 return decode_bytes(self.output_bytes, self.encoding) 48 49 @property 50 def error(self) -> str: 51 "Error from command as a string (if it is not redirected)." 52 return decode_bytes(self.error_bytes, self.encoding) 53 54 @property 55 def exit_signal(self) -> signal.Signals | None: 56 "Signal that caused the command to exit, or None if no signal." 57 if self.exit_code >= 0: 58 return None 59 return signal.Signals(-self.exit_code) 60 61 def __bool__(self) -> bool: 62 "Return true if exit_code is 0." 63 return self.exit_code == 0
Concrete class for the result of a Command.
output: str
44 @property 45 def output(self) -> str: 46 "Output of command as a string." 47 return decode_bytes(self.output_bytes, self.encoding)
Output of command as a string.
error: str
49 @property 50 def error(self) -> str: 51 "Error from command as a string (if it is not redirected)." 52 return decode_bytes(self.error_bytes, self.encoding)
Error from command as a string (if it is not redirected).
exit_signal: signal.Signals | None
54 @property 55 def exit_signal(self) -> signal.Signals | None: 56 "Signal that caused the command to exit, or None if no signal." 57 if self.exit_code >= 0: 58 return None 59 return signal.Signals(-self.exit_code)
Signal that caused the command to exit, or None if no signal.
class
ResultError(builtins.Exception):
13class ResultError(Exception): 14 "Represents a non-zero exit status." 15 16 @property 17 def result(self) -> "shellous.Result": 18 "Result of the command." 19 return self.args[0]
Represents a non-zero exit status.
class
Runner:
420class Runner: 421 """Runner is an asynchronous context manager that runs a command. 422 423 ``` 424 async with Runner(cmd) as run: 425 # process streams: run.stdin, run.stdout, run.stderr (if not None) 426 result = run.result() 427 ``` 428 """ 429 430 stdin: asyncio.StreamWriter | None = None 431 "Process standard input." 432 433 stdout: asyncio.StreamReader | None = None 434 "Process standard output." 435 436 stderr: asyncio.StreamReader | None = None 437 "Process standard error." 438 439 _options: _RunOptions 440 _tasks: list[asyncio.Task[Any]] 441 _proc: "asyncio.subprocess.Process | None" = None 442 _cancelled: bool = False 443 _timer: asyncio.TimerHandle | None = None 444 _timed_out: bool = False 445 _last_signal: int | None = None 446 _ignore_cancel_signal: bool = False 447 448 def __init__(self, command: "shellous.Command[Any]"): 449 self._options = _RunOptions(command) 450 self._tasks = [] 451 452 @property 453 def name(self) -> str: 454 "Return name of process being run." 455 return self.command.name 456 457 @property 458 def options(self) -> "shellous.Options": 459 "Return options for process being run." 460 return self.command.options 461 462 @property 463 def command(self) -> "shellous.Command[Any]": 464 "Return the command being run." 465 return self._options.command 466 467 @property 468 def pid(self) -> int | None: 469 "Return the command's process ID." 470 if not self._proc: 471 return None 472 return self._proc.pid 473 474 @property 475 def returncode(self) -> int | None: 476 "Process's exit code." 477 if not self._proc: 478 if self._cancelled: 479 # The process was cancelled before starting. 480 return CANCELLED_EXIT_CODE 481 return None 482 code = self._proc.returncode 483 if code == _UNKNOWN_EXIT_CODE and self._last_signal is not None: 484 # After sending a signal, `waitpid` may fail to locate the child 485 # process. In this case, map the status to the last signal we sent. 486 # For more on this, see https://github.com/python/cpython/issues/87744 487 return -self._last_signal # pylint: disable=invalid-unary-operand-type 488 return code 489 490 @property 491 def cancelled(self) -> bool: 492 "Return True if the command was cancelled." 493 return self._cancelled 494 495 @property 496 def pty_fd(self) -> int | None: 497 """The file descriptor used to communicate with the child PTY process. 498 499 Returns None if the process is not using a PTY. 500 """ 501 pty_fds = self._options.pty_fds 502 if pty_fds is not None: 503 return pty_fds.parent_fd 504 return None 505 506 @property 507 def pty_eof(self) -> bytes | None: 508 """Byte sequence used to indicate EOF when written to the PTY child. 509 510 Returns None if process is not using a PTY. 511 """ 512 pty_fds = self._options.pty_fds 513 if pty_fds is not None: 514 return pty_fds.eof 515 return None 516 517 def result(self, *, check: bool = True) -> Result: 518 "Check process exit code and raise a ResultError if necessary." 519 code = self.returncode 520 if code is None: 521 raise TypeError("Runner.result(): Process has not exited") 522 523 if self._ignore_cancel_signal and not self._cancelled: 524 # Check if we need to replace a non-zero exit code for an early 525 # terminated process with zero. (See `OutputInterrupted`) 526 cancel_signal = self.command.options.cancel_signal 527 if cancel_signal is not None: 528 code = _map_graceful_exit_code(code, cancel_signal) 529 530 result = Result( 531 exit_code=code, 532 output_bytes=bytes(self._options.output_bytes or b""), 533 error_bytes=bytes(self._options.error_bytes or b""), 534 cancelled=self._cancelled, 535 encoding=self._options.encoding, 536 ) 537 538 if not check: 539 return result 540 541 return check_result( 542 result, 543 self.command.options, 544 self._cancelled, 545 self._timed_out, 546 ) 547 548 def add_task( 549 self, 550 coro: Coroutine[Any, Any, _T], 551 tag: str = "", 552 ) -> asyncio.Task[_T]: 553 "Add a background task." 554 task_name = f"{self.name}#{tag}" 555 task = asyncio.create_task(coro, name=task_name) 556 self._tasks.append(task) 557 return task 558 559 def send_signal(self, sig: int) -> None: 560 "Send an arbitrary signal to the process if it is running." 561 if self.returncode is None: 562 self._signal(sig) 563 564 def cancel(self) -> None: 565 "Cancel the running process if it is running." 566 if self.returncode is None: 567 self._signal(self.command.options.cancel_signal) 568 569 def _is_bsd_pty(self) -> bool: 570 "Return true if we're running a pty on BSD." 571 return BSD_DERIVED and bool(self._options.pty_fds) 572 573 @log_method(LOG_DETAIL) 574 async def _wait(self) -> None: 575 "Normal wait for background I/O tasks and process to finish." 576 assert self._proc 577 578 try: 579 if self._tasks: 580 await harvest(*self._tasks, trustee=self) 581 if self._is_bsd_pty(): 582 await self._waiter() 583 584 except asyncio.CancelledError: 585 LOGGER.debug("Runner.wait cancelled %r", self) 586 self._set_cancelled() 587 self._tasks.clear() # all tasks were cancelled 588 await self._kill() 589 590 except OutputInterrupted: 591 LOGGER.debug("Runner.wait output interrupted %r", self) 592 self._tasks.clear() # all tasks were cancelled 593 # Abort process but map the negative exit_code from the cancel 594 # signal to 0. 595 self._ignore_cancel_signal = True 596 await self._kill() 597 598 except Exception as ex: 599 LOGGER.debug("Runner.wait exited with error %r ex=%r", self, ex) 600 self._tasks.clear() # all tasks were cancelled 601 await self._kill() 602 raise # re-raise exception 603 604 @log_method(LOG_DETAIL) 605 async def _wait_pid(self): 606 "Manually poll `waitpid` until process finishes." 607 assert self._is_bsd_pty() 608 609 while True: 610 assert self._proc is not None # (pyright) 611 612 if poll_wait_pid(self._proc): 613 break 614 await asyncio.sleep(0.025) 615 616 @log_method(LOG_DETAIL) 617 async def _kill(self): 618 "Kill process and wait for it to finish." 619 assert self._proc 620 621 cancel_timeout = self.command.options.cancel_timeout 622 cancel_signal = self.command.options.cancel_signal 623 624 try: 625 # If not already done, send cancel signal. 626 if self._proc.returncode is None: 627 self._signal(cancel_signal) 628 629 if self._tasks: 630 await harvest(*self._tasks, timeout=cancel_timeout, trustee=self) 631 632 if self._proc.returncode is None: 633 await harvest(self._waiter(), timeout=cancel_timeout, trustee=self) 634 635 except (asyncio.CancelledError, asyncio.TimeoutError) as ex: 636 LOGGER.warning("Runner.kill %r (ex)=%r", self, ex) 637 if _is_cancelled(ex): 638 self._set_cancelled() 639 await self._kill_wait() 640 641 except (Exception, GeneratorExit) as ex: 642 LOGGER.warning("Runner.kill %r ex=%r", self, ex) 643 await self._kill_wait() 644 raise 645 646 def _signal(self, sig: int | None): 647 "Send a signal to the process." 648 assert self._proc is not None # (pyright) 649 650 if LOG_DETAIL: 651 LOGGER.debug("Runner.signal %r signal=%r", self, sig) 652 self._audit_callback("signal", signal=sig) 653 654 if sig is None: 655 self._proc.kill() 656 else: 657 self._proc.send_signal(sig) 658 self._last_signal = sig 659 660 @log_method(LOG_DETAIL) 661 async def _kill_wait(self): 662 "Wait for killed process to exit." 663 assert self._proc 664 665 # Check if process is already done. 666 if self._proc.returncode is not None: 667 return 668 669 try: 670 self._signal(None) 671 await harvest(self._waiter(), timeout=_KILL_TIMEOUT, trustee=self) 672 except asyncio.TimeoutError as ex: 673 # Manually check if the process is still running. 674 if poll_wait_pid(self._proc): 675 LOGGER.warning("%r process reaped manually %r", self, self._proc) 676 else: 677 LOGGER.error("%r failed to kill process %r", self, self._proc) 678 raise RuntimeError(f"Unable to kill process {self._proc!r}") from ex 679 680 @log_method(LOG_DETAIL) 681 async def __aenter__(self): 682 "Set up redirections and launch subprocess." 683 self._audit_callback("start") 684 try: 685 return await self._start() 686 except BaseException as ex: 687 self._stop_timer() # failsafe just in case 688 self._audit_callback("stop", failure=type(ex).__name__) 689 raise 690 finally: 691 if self._cancelled and self.command.options._catch_cancelled_error: 692 # Raises ResultError instead of CancelledError. 693 self.result() 694 695 @log_method(LOG_DETAIL) 696 async def _start(self): 697 "Set up redirections and launch subprocess." 698 # assert self._proc is None 699 assert not self._tasks 700 701 try: 702 # Set up subprocess arguments and launch subprocess. 703 with self._options as opts: 704 await self._subprocess_spawn(opts) 705 706 assert self._proc is not None 707 stdin = self._proc.stdin 708 stdout = self._proc.stdout 709 stderr = self._proc.stderr 710 711 # Assign pty streams. 712 if opts.pty_fds: 713 assert (stdin, stdout) == (None, None) 714 stdin, stdout = opts.pty_fds.writer, opts.pty_fds.reader 715 716 if stderr is not None: 717 limit = None 718 if opts.error_bytes is not None: 719 error = opts.error_bytes 720 limit = opts.command.options.error_limit 721 elif opts.is_stderr_only: 722 assert stdout is None 723 assert opts.output_bytes is not None 724 error = opts.output_bytes 725 else: 726 error = opts.command.options.error 727 stderr = self._setup_output_sink( 728 stderr, error, opts.encoding, "stderr", limit 729 ) 730 731 if stdout is not None: 732 limit = None 733 if opts.output_bytes is not None: 734 output = opts.output_bytes 735 else: 736 output = opts.command.options.output 737 stdout = self._setup_output_sink( 738 stdout, output, opts.encoding, "stdout", limit 739 ) 740 741 if stdin is not None: 742 stdin = self._setup_input_source(stdin, opts) 743 744 except (Exception, asyncio.CancelledError) as ex: 745 LOGGER.debug("Runner._start %r ex=%r", self, ex) 746 if _is_cancelled(ex): 747 self._set_cancelled() 748 if self._proc: 749 await self._kill() 750 if self._options.pty_fds: 751 self._options.pty_fds.close() 752 raise 753 754 # Make final streams available. These may be different from `self.proc` 755 # versions. 756 self.stdin = stdin 757 self.stdout = stdout 758 self.stderr = stderr 759 760 # Add a task to monitor for when the process finishes. 761 if not self._is_bsd_pty(): 762 self.add_task(self._waiter(), "waiter") 763 764 # Set a timer to cancel the current task after a timeout. 765 self._start_timer(self.command.options.timeout) 766 767 return self 768 769 @log_method(LOG_DETAIL) 770 async def _subprocess_spawn(self, opts: _RunOptions): 771 "Start the subprocess." 772 assert self._proc is None 773 774 # Second half of pty setup. 775 if opts.pty_fds: 776 opts.pty_fds = await opts.pty_fds.open_streams() 777 778 # Check for task cancellation and yield right before exec'ing. If the 779 # current task is already cancelled, this will raise a CancelledError, 780 # and we save ourselves the work of launching and immediately killing 781 # a process. 782 await asyncio.sleep(0) 783 784 # Launch the subprocess (always completes even if cancelled). 785 await uninterrupted(self._subprocess_exec(opts)) 786 787 # Launch the process substitution commands (if any). 788 for cmd in opts.subcmds: 789 self.add_task(cmd.coro(), "procsub") 790 791 @log_method(LOG_DETAIL) 792 async def _subprocess_exec(self, opts: _RunOptions): 793 "Start the subprocess and assign to `self.proc`." 794 with log_timer("asyncio.create_subprocess_exec"): 795 sys.audit(EVENT_SHELLOUS_EXEC, opts.pos_args[0]) 796 with pty_util.set_ignore_child_watcher( 797 BSD_DERIVED and opts.pty_fds is not None 798 ): 799 self._proc = await asyncio.create_subprocess_exec( 800 *opts.pos_args, 801 **opts.kwd_args, 802 ) 803 804 @log_method(LOG_DETAIL) 805 async def _waiter(self): 806 "Run task that waits for process to exit." 807 assert self._proc is not None # (pyright) 808 809 try: 810 if self._is_bsd_pty(): 811 await self._wait_pid() 812 else: 813 await self._proc.wait() 814 finally: 815 self._stop_timer() 816 817 def _set_cancelled(self): 818 "Set the cancelled flag, and cancel any inflight timers." 819 self._cancelled = True 820 self._stop_timer() 821 822 def _start_timer(self, timeout: float | None): 823 "Start an optional timer to cancel the process if `timeout` desired." 824 assert self._timer is None 825 if timeout is not None: 826 loop = asyncio.get_running_loop() 827 task = asyncio.current_task() 828 assert task is not None 829 self._timer = loop.call_later( 830 timeout, 831 self._set_timer_expired, 832 task, 833 ) 834 835 def _set_timer_expired(self, main_task: asyncio.Task[Any]): 836 "Set a flag when the timer expires and cancel the main task." 837 self._timed_out = True 838 self._timer = None 839 main_task.cancel() 840 841 def _stop_timer(self): 842 if self._timer: 843 self._timer.cancel() 844 self._timer = None 845 846 def _setup_input_source( 847 self, 848 stream: asyncio.StreamWriter, 849 opts: _RunOptions, 850 ): 851 "Set up a task to read from custom input source." 852 tag = "stdin" 853 eof = opts.pty_fds.eof if opts.pty_fds else None 854 855 if opts.input_bytes is not None: 856 self.add_task(redir.write_stream(opts.input_bytes, stream, eof), tag) 857 return None 858 859 source = opts.command.options.input 860 861 if isinstance(source, asyncio.StreamReader): 862 self.add_task(redir.write_reader(source, stream, eof), tag) 863 return None 864 865 if isinstance(source, io.BytesIO): 866 self.add_task(redir.write_stream(source.getvalue(), stream, eof), tag) 867 return None 868 869 if isinstance(source, io.StringIO): 870 input_bytes = encode_bytes(source.getvalue(), opts.encoding) 871 self.add_task(redir.write_stream(input_bytes, stream, eof), tag) 872 return None 873 874 if isinstance(source, cabc.AsyncGenerator): 875 obj = cast(AsyncGenerator[Any, Any], source) 876 if _is_async_gen_closed(obj): 877 LOGGER.warning("Runner: Async generator input is closed: %r", obj) 878 raise ValueError(f"Async generator input is closed: {source!r}") 879 self.add_task(redir.write_asyncgen(obj, stream, opts.encoding, eof), tag) 880 return None 881 882 return stream 883 884 def _setup_output_sink( 885 self, 886 stream: asyncio.StreamReader, 887 sink: Any, 888 encoding: str, 889 tag: str, 890 limit: int | None = None, 891 ) -> asyncio.StreamReader | None: 892 "Set up a task to write to custom output sink." 893 if isinstance(sink, io.StringIO): 894 self.add_task(redir.copy_stringio(stream, sink, encoding), tag) 895 return None 896 897 if isinstance(sink, io.BytesIO): 898 self.add_task(redir.copy_bytesio(stream, sink), tag) 899 return None 900 901 if isinstance(sink, bytearray): 902 # N.B. `limit` is only supported for bytearray. 903 if limit is not None: 904 self.add_task(redir.copy_bytearray_limit(stream, sink, limit), tag) 905 else: 906 self.add_task(redir.copy_bytearray(stream, sink), tag) 907 return None 908 909 if isinstance(sink, Logger): 910 self.add_task(redir.copy_logger(stream, sink, encoding), tag) 911 return None 912 913 if isinstance(sink, asyncio.StreamWriter): 914 self.add_task(redir.copy_streamwriter(stream, sink), tag) 915 return None 916 917 if isinstance(sink, cabc.AsyncGenerator): 918 obj = cast(AsyncGenerator[Any, Any], sink) 919 if _is_async_gen_closed(obj): 920 LOGGER.warning("Runner: Async generator output is closed: %r", obj) 921 raise ValueError(f"Async generator output is closed: {sink!r}") 922 self.add_task(redir.copy_asyncgen(stream, sink), tag) 923 return None 924 925 return stream 926 927 @log_method(LOG_DETAIL) 928 async def __aexit__( 929 self, 930 _exc_type: type[BaseException] | None, 931 exc_value: BaseException | None, 932 _exc_tb: TracebackType | None, 933 ): 934 "Wait for process to exit and handle cancellation." 935 suppress = False 936 try: 937 suppress = await self._finish(exc_value) 938 except asyncio.CancelledError: 939 LOGGER.debug("Runner cancelled inside _finish %r", self) 940 self._set_cancelled() 941 finally: 942 self._stop_timer() # failsafe just in case 943 self._audit_callback("stop") 944 # If `timeout` expired, raise TimeoutError rather than CancelledError. 945 if ( 946 self._cancelled 947 and self._timed_out 948 and not self.command.options._catch_cancelled_error 949 ): 950 raise asyncio.TimeoutError() 951 return suppress 952 953 @log_method(LOG_DETAIL) 954 async def _finish(self, exc_value: BaseException | None): 955 "Finish the run. Return True only if `exc_value` should be suppressed." 956 assert self._proc 957 958 try: 959 if exc_value is not None: 960 if _is_cancelled(exc_value): 961 self._set_cancelled() 962 await self._kill() 963 return self._cancelled 964 965 await self._wait() 966 return False 967 968 finally: 969 await self._close() 970 971 @log_method(LOG_DETAIL) 972 async def _close(self): 973 "Make sure that our resources are properly closed." 974 assert self._proc is not None 975 976 if self._options.pty_fds: 977 self._options.pty_fds.close() 978 979 # Make sure the transport is closed (for asyncio and uvloop). 980 self._proc._transport.close() # pyright: ignore 981 982 # _close can be called when unwinding exceptions. We need to handle 983 # the case that the process has not exited yet. 984 if self._proc.returncode is None: 985 LOGGER.critical("Runner._close process still running %r", self._proc) 986 return 987 988 try: 989 # Make sure that original stdin is properly closed. `wait_closed` 990 # will raise a BrokenPipeError if not all input was properly written. 991 if self._proc.stdin is not None: 992 self._proc.stdin.close() 993 await harvest( 994 self._proc.stdin.wait_closed(), 995 timeout=_CLOSE_TIMEOUT, 996 cancel_finish=True, # finish `wait_closed` if cancelled 997 trustee=self, 998 ) 999 1000 except asyncio.TimeoutError: 1001 LOGGER.critical("Runner._close %r timeout stdin=%r", self, self._proc.stdin) 1002 1003 def _audit_callback( 1004 self, 1005 phase: str, 1006 *, 1007 failure: str = "", 1008 signal: int | None = None, 1009 ): 1010 "Call `audit_callback` if there is one." 1011 callback = self.command.options.audit_callback 1012 if callback: 1013 sig = _signame(signal) if phase == "signal" else "" 1014 info: shellous.AuditEventInfo = { 1015 "runner": self, 1016 "failure": failure, 1017 "signal": sig, 1018 } 1019 callback(phase, info) 1020 1021 def __repr__(self) -> str: 1022 "Return string representation of Runner." 1023 cancelled = " cancelled" if self._cancelled else "" 1024 if self._proc: 1025 procinfo = f" pid={self._proc.pid} exit_code={self.returncode}" 1026 else: 1027 procinfo = " pid=None" 1028 return f"<Runner {self.name!r}{cancelled}{procinfo}>" 1029 1030 async def _readlines(self): 1031 "Iterate over lines in stdout/stderr" 1032 stream = self.stdout or self.stderr 1033 if stream: 1034 async for line in redir.read_lines(stream, self._options.encoding): 1035 yield line 1036 1037 def __aiter__(self) -> AsyncIterator[str]: 1038 "Return asynchronous iterator over stdout/stderr." 1039 return self._readlines() 1040 1041 @staticmethod 1042 async def run_command( 1043 command: "shellous.Command[Any]", 1044 *, 1045 _run_future: asyncio.Future["Runner"] | None = None, 1046 ) -> str | Result: 1047 "Run a command. This is the main entry point for Runner." 1048 if not _run_future and _is_multiple_capture(command): 1049 LOGGER.warning("run_command: multiple capture requires 'async with'") 1050 _cleanup(command) 1051 raise ValueError("multiple capture requires 'async with'") 1052 1053 async with Runner(command) as run: 1054 if _run_future is not None: 1055 # Return streams to caller in another task. 1056 _run_future.set_result(run) 1057 1058 result = run.result() 1059 if command.options._return_result: 1060 return result 1061 return result.output
Runner is an asynchronous context manager that runs a command.
async with Runner(cmd) as run:
# process streams: run.stdin, run.stdout, run.stderr (if not None)
result = run.result()
Runner(command: Command[typing.Any])
name: str
452 @property 453 def name(self) -> str: 454 "Return name of process being run." 455 return self.command.name
Return name of process being run.
options: Options
457 @property 458 def options(self) -> "shellous.Options": 459 "Return options for process being run." 460 return self.command.options
Return options for process being run.
command: Command[typing.Any]
462 @property 463 def command(self) -> "shellous.Command[Any]": 464 "Return the command being run." 465 return self._options.command
Return the command being run.
pid: int | None
467 @property 468 def pid(self) -> int | None: 469 "Return the command's process ID." 470 if not self._proc: 471 return None 472 return self._proc.pid
Return the command's process ID.
returncode: int | None
474 @property 475 def returncode(self) -> int | None: 476 "Process's exit code." 477 if not self._proc: 478 if self._cancelled: 479 # The process was cancelled before starting. 480 return CANCELLED_EXIT_CODE 481 return None 482 code = self._proc.returncode 483 if code == _UNKNOWN_EXIT_CODE and self._last_signal is not None: 484 # After sending a signal, `waitpid` may fail to locate the child 485 # process. In this case, map the status to the last signal we sent. 486 # For more on this, see https://github.com/python/cpython/issues/87744 487 return -self._last_signal # pylint: disable=invalid-unary-operand-type 488 return code
Process's exit code.
cancelled: bool
490 @property 491 def cancelled(self) -> bool: 492 "Return True if the command was cancelled." 493 return self._cancelled
Return True if the command was cancelled.
pty_fd: int | None
495 @property 496 def pty_fd(self) -> int | None: 497 """The file descriptor used to communicate with the child PTY process. 498 499 Returns None if the process is not using a PTY. 500 """ 501 pty_fds = self._options.pty_fds 502 if pty_fds is not None: 503 return pty_fds.parent_fd 504 return None
The file descriptor used to communicate with the child PTY process.
Returns None if the process is not using a PTY.
pty_eof: bytes | None
506 @property 507 def pty_eof(self) -> bytes | None: 508 """Byte sequence used to indicate EOF when written to the PTY child. 509 510 Returns None if process is not using a PTY. 511 """ 512 pty_fds = self._options.pty_fds 513 if pty_fds is not None: 514 return pty_fds.eof 515 return None
Byte sequence used to indicate EOF when written to the PTY child.
Returns None if process is not using a PTY.
517 def result(self, *, check: bool = True) -> Result: 518 "Check process exit code and raise a ResultError if necessary." 519 code = self.returncode 520 if code is None: 521 raise TypeError("Runner.result(): Process has not exited") 522 523 if self._ignore_cancel_signal and not self._cancelled: 524 # Check if we need to replace a non-zero exit code for an early 525 # terminated process with zero. (See `OutputInterrupted`) 526 cancel_signal = self.command.options.cancel_signal 527 if cancel_signal is not None: 528 code = _map_graceful_exit_code(code, cancel_signal) 529 530 result = Result( 531 exit_code=code, 532 output_bytes=bytes(self._options.output_bytes or b""), 533 error_bytes=bytes(self._options.error_bytes or b""), 534 cancelled=self._cancelled, 535 encoding=self._options.encoding, 536 ) 537 538 if not check: 539 return result 540 541 return check_result( 542 result, 543 self.command.options, 544 self._cancelled, 545 self._timed_out, 546 )
Check process exit code and raise a ResultError if necessary.
def
add_task( self, coro: Coroutine[Any, Any, ~_T], tag: str = '') -> _asyncio.Task[~_T]:
548 def add_task( 549 self, 550 coro: Coroutine[Any, Any, _T], 551 tag: str = "", 552 ) -> asyncio.Task[_T]: 553 "Add a background task." 554 task_name = f"{self.name}#{tag}" 555 task = asyncio.create_task(coro, name=task_name) 556 self._tasks.append(task) 557 return task
Add a background task.
def
send_signal(self, sig: int) -> None:
559 def send_signal(self, sig: int) -> None: 560 "Send an arbitrary signal to the process if it is running." 561 if self.returncode is None: 562 self._signal(sig)
Send an arbitrary signal to the process if it is running.
def
cancel(self) -> None:
564 def cancel(self) -> None: 565 "Cancel the running process if it is running." 566 if self.returncode is None: 567 self._signal(self.command.options.cancel_signal)
Cancel the running process if it is running.
@log_method(LOG_DETAIL)
async def
__aenter__(self):
680 @log_method(LOG_DETAIL) 681 async def __aenter__(self): 682 "Set up redirections and launch subprocess." 683 self._audit_callback("start") 684 try: 685 return await self._start() 686 except BaseException as ex: 687 self._stop_timer() # failsafe just in case 688 self._audit_callback("stop", failure=type(ex).__name__) 689 raise 690 finally: 691 if self._cancelled and self.command.options._catch_cancelled_error: 692 # Raises ResultError instead of CancelledError. 693 self.result()
Set up redirections and launch subprocess.
def
__aiter__(self) -> AsyncIterator[str]:
1037 def __aiter__(self) -> AsyncIterator[str]: 1038 "Return asynchronous iterator over stdout/stderr." 1039 return self._readlines()
Return asynchronous iterator over stdout/stderr.
@staticmethod
async def
run_command( command: Command[typing.Any], *, _run_future: _asyncio.Future[Runner] | None = None) -> str | Result:
1041 @staticmethod 1042 async def run_command( 1043 command: "shellous.Command[Any]", 1044 *, 1045 _run_future: asyncio.Future["Runner"] | None = None, 1046 ) -> str | Result: 1047 "Run a command. This is the main entry point for Runner." 1048 if not _run_future and _is_multiple_capture(command): 1049 LOGGER.warning("run_command: multiple capture requires 'async with'") 1050 _cleanup(command) 1051 raise ValueError("multiple capture requires 'async with'") 1052 1053 async with Runner(command) as run: 1054 if _run_future is not None: 1055 # Return streams to caller in another task. 1056 _run_future.set_result(run) 1057 1058 result = run.result() 1059 if command.options._return_result: 1060 return result 1061 return result.output
Run a command. This is the main entry point for Runner.
class
PipeRunner:
1064class PipeRunner: 1065 """PipeRunner is an asynchronous context manager that runs a pipeline. 1066 1067 ``` 1068 async with pipe.run() as run: 1069 # process run.stdin, run.stdout, run.stderr (if not None) 1070 result = run.result() 1071 ``` 1072 """ 1073 1074 stdin: asyncio.StreamWriter | None = None 1075 "Pipeline standard input." 1076 1077 stdout: asyncio.StreamReader | None = None 1078 "Pipeline standard output." 1079 1080 stderr: asyncio.StreamReader | None = None 1081 "Pipeline standard error." 1082 1083 _pipe: "shellous.Pipeline[Any]" 1084 _capturing: bool 1085 _tasks: list[asyncio.Task[Any]] 1086 _encoding: str 1087 _cancelled: bool = False 1088 _results: list[BaseException | Result] | None = None 1089 _pid: int = -1 1090 1091 def __init__(self, pipe: "shellous.Pipeline[Any]", *, capturing: bool): 1092 """`capturing=True` indicates we are within an `async with` block and 1093 client needs to access `stdin` and `stderr` streams. 1094 """ 1095 assert len(pipe.commands) > 1 1096 1097 self._pipe = pipe 1098 self._cancelled = False 1099 self._tasks = [] 1100 self._capturing = capturing 1101 self._encoding = pipe.options.encoding 1102 1103 @property 1104 def name(self) -> str: 1105 "Return name of the pipeline." 1106 return self._pipe.name 1107 1108 @property 1109 def options(self) -> "shellous.Options": 1110 """Return options for pipeline being run. 1111 1112 These are the options for the last command in the pipeline. 1113 """ 1114 return self._pipe.options 1115 1116 @property 1117 def pid(self) -> int | None: 1118 """Return the process ID for the first command in the pipeline. 1119 1120 The PID is only available when `capturing=True`. 1121 """ 1122 if self._pid < 0: 1123 return None 1124 return self._pid 1125 1126 def result(self, *, check: bool = True) -> Result: 1127 "Return `Result` object for PipeRunner." 1128 assert self._results is not None 1129 1130 result = convert_result_list(self._results, self._cancelled) 1131 if not check: 1132 return result 1133 1134 return check_result(result, self._pipe.options, self._cancelled) 1135 1136 def add_task( 1137 self, 1138 coro: Coroutine[Any, Any, _T], 1139 tag: str = "", 1140 ) -> asyncio.Task[_T]: 1141 "Add a background task." 1142 task_name = f"{self.name}#{tag}" 1143 task = asyncio.create_task(coro, name=task_name) 1144 self._tasks.append(task) 1145 return task 1146 1147 @log_method(LOG_DETAIL) 1148 async def _wait(self, *, kill: bool = False): 1149 "Wait for pipeline to finish." 1150 assert self._results is None 1151 1152 if kill: 1153 LOGGER.debug("PipeRunner.wait killing pipe %r", self) 1154 for task in self._tasks: 1155 task.cancel() 1156 1157 cancelled, self._results = await harvest_results(*self._tasks, trustee=self) 1158 if cancelled: 1159 self._cancelled = True 1160 self._tasks.clear() # clear all tasks when done 1161 1162 @log_method(LOG_DETAIL) 1163 async def __aenter__(self) -> "PipeRunner": 1164 "Set up redirections and launch pipeline." 1165 try: 1166 return await self._start() 1167 except (Exception, asyncio.CancelledError) as ex: 1168 LOGGER.warning("PipeRunner enter %r ex=%r", self, ex) 1169 if _is_cancelled(ex): 1170 self._cancelled = True 1171 await self._wait(kill=True) 1172 raise 1173 1174 @log_method(LOG_DETAIL) 1175 async def __aexit__( 1176 self, 1177 _exc_type: type[BaseException] | None, 1178 exc_value: BaseException | None, 1179 _exc_tb: TracebackType | None, 1180 ): 1181 "Wait for pipeline to exit and handle cancellation." 1182 suppress = False 1183 try: 1184 suppress = await self._finish(exc_value) 1185 except asyncio.CancelledError: 1186 LOGGER.warning("PipeRunner cancelled inside _finish %r", self) 1187 self._cancelled = True 1188 return suppress 1189 1190 @log_method(LOG_DETAIL) 1191 async def _finish(self, exc_value: BaseException | None) -> bool: 1192 "Wait for pipeline to exit and handle cancellation." 1193 if exc_value is not None: 1194 LOGGER.warning("PipeRunner._finish exc_value=%r", exc_value) 1195 if _is_cancelled(exc_value): 1196 self._cancelled = True 1197 await self._wait(kill=True) 1198 return self._cancelled 1199 1200 await self._wait() 1201 return False 1202 1203 @log_method(LOG_DETAIL) 1204 async def _start(self): 1205 "Set up redirection and launch pipeline." 1206 open_fds: list[int] = [] 1207 1208 try: 1209 stdin = None 1210 stdout = None 1211 stderr = None 1212 1213 cmds = self._setup_pipeline(open_fds) 1214 1215 if self._capturing: 1216 stdin, stdout, stderr = await self._setup_capturing(cmds) 1217 else: 1218 for cmd in cmds: 1219 self.add_task(cmd.coro()) 1220 1221 self.stdin = stdin 1222 self.stdout = stdout 1223 self.stderr = stderr 1224 1225 return self 1226 1227 except BaseException: # pylint: disable=broad-except 1228 # Clean up after any exception *including* CancelledError. 1229 close_fds(open_fds) 1230 raise 1231 1232 def _setup_pipeline(self, open_fds: list[int]): 1233 """Return the pipeline stitched together with pipe fd's. 1234 1235 Each created open file descriptor is added to `open_fds` so it can 1236 be closed if there's an exception later. 1237 """ 1238 cmds = list(self._pipe.commands) 1239 1240 cmd_count = len(cmds) 1241 for i in range(cmd_count - 1): 1242 read_fd, write_fd = os.pipe() 1243 open_fds.extend((read_fd, write_fd)) 1244 1245 cmds[i] = cmds[i].stdout(write_fd, close=True) 1246 cmds[i + 1] = cmds[i + 1].stdin(read_fd, close=True) 1247 1248 for i in range(cmd_count): 1249 cmds[i] = cmds[i].set(_return_result=True, _catch_cancelled_error=True) 1250 1251 return cmds 1252 1253 @log_method(LOG_DETAIL) 1254 async def _setup_capturing(self, cmds: "list[shellous.Command[Any]]"): 1255 """Set up capturing and return (stdin, stdout, stderr) streams.""" 1256 loop = asyncio.get_event_loop() 1257 first_fut = loop.create_future() 1258 last_fut = loop.create_future() 1259 1260 first_coro = cmds[0].coro(_run_future=first_fut) 1261 last_coro = cmds[-1].coro(_run_future=last_fut) 1262 middle_coros = [cmd.coro() for cmd in cmds[1:-1]] 1263 1264 # Tag each task name with the index of the command in the pipe. 1265 self.add_task(first_coro, "0") 1266 for i, coro in enumerate(middle_coros): 1267 self.add_task(coro, str(i + 1)) 1268 self.add_task(last_coro, str(len(cmds) - 1)) 1269 1270 # When capturing, we need the first and last commands in the 1271 # pipe to signal when they are ready. 1272 first_ready, last_ready = await asyncio.gather(first_fut, last_fut) 1273 1274 stdin, stdout, stderr = ( 1275 first_ready.stdin, 1276 last_ready.stdout, 1277 last_ready.stderr, 1278 ) 1279 self._pid = first_ready.pid 1280 1281 return (stdin, stdout, stderr) 1282 1283 def __repr__(self) -> str: 1284 "Return string representation of PipeRunner." 1285 cancelled_info = "" 1286 if self._cancelled: 1287 cancelled_info = " cancelled" 1288 result_info = "" 1289 if self._results: 1290 result_info = f" results={self._results!r}" 1291 return f"<PipeRunner {self.name!r}{cancelled_info}{result_info}>" 1292 1293 async def _readlines(self) -> AsyncIterator[str]: 1294 "Iterate over lines in stdout/stderr" 1295 stream = self.stdout or self.stderr 1296 if stream: 1297 async for line in redir.read_lines(stream, self._encoding): 1298 yield line 1299 1300 def __aiter__(self) -> AsyncIterator[str]: 1301 "Return asynchronous iterator over stdout/stderr." 1302 return self._readlines() 1303 1304 @staticmethod 1305 async def run_pipeline(pipe: "shellous.Pipeline[Any]") -> str | Result: 1306 "Run a pipeline. This is the main entry point for PipeRunner." 1307 run = PipeRunner(pipe, capturing=False) 1308 async with run: 1309 pass 1310 1311 result = run.result() 1312 if pipe.options._return_result: 1313 return result 1314 return result.output
PipeRunner is an asynchronous context manager that runs a pipeline.
async with pipe.run() as run:
# process run.stdin, run.stdout, run.stderr (if not None)
result = run.result()
PipeRunner(pipe: Pipeline[typing.Any], *, capturing: bool)
1091 def __init__(self, pipe: "shellous.Pipeline[Any]", *, capturing: bool): 1092 """`capturing=True` indicates we are within an `async with` block and 1093 client needs to access `stdin` and `stderr` streams. 1094 """ 1095 assert len(pipe.commands) > 1 1096 1097 self._pipe = pipe 1098 self._cancelled = False 1099 self._tasks = [] 1100 self._capturing = capturing 1101 self._encoding = pipe.options.encoding
name: str
1103 @property 1104 def name(self) -> str: 1105 "Return name of the pipeline." 1106 return self._pipe.name
Return name of the pipeline.
options: Options
1108 @property 1109 def options(self) -> "shellous.Options": 1110 """Return options for pipeline being run. 1111 1112 These are the options for the last command in the pipeline. 1113 """ 1114 return self._pipe.options
Return options for pipeline being run.
These are the options for the last command in the pipeline.
pid: int | None
1116 @property 1117 def pid(self) -> int | None: 1118 """Return the process ID for the first command in the pipeline. 1119 1120 The PID is only available when `capturing=True`. 1121 """ 1122 if self._pid < 0: 1123 return None 1124 return self._pid
Return the process ID for the first command in the pipeline.
The PID is only available when capturing=True.
1126 def result(self, *, check: bool = True) -> Result: 1127 "Return `Result` object for PipeRunner." 1128 assert self._results is not None 1129 1130 result = convert_result_list(self._results, self._cancelled) 1131 if not check: 1132 return result 1133 1134 return check_result(result, self._pipe.options, self._cancelled)
Return Result object for PipeRunner.
def
add_task( self, coro: Coroutine[Any, Any, ~_T], tag: str = '') -> _asyncio.Task[~_T]:
1136 def add_task( 1137 self, 1138 coro: Coroutine[Any, Any, _T], 1139 tag: str = "", 1140 ) -> asyncio.Task[_T]: 1141 "Add a background task." 1142 task_name = f"{self.name}#{tag}" 1143 task = asyncio.create_task(coro, name=task_name) 1144 self._tasks.append(task) 1145 return task
Add a background task.
1162 @log_method(LOG_DETAIL) 1163 async def __aenter__(self) -> "PipeRunner": 1164 "Set up redirections and launch pipeline." 1165 try: 1166 return await self._start() 1167 except (Exception, asyncio.CancelledError) as ex: 1168 LOGGER.warning("PipeRunner enter %r ex=%r", self, ex) 1169 if _is_cancelled(ex): 1170 self._cancelled = True 1171 await self._wait(kill=True) 1172 raise
Set up redirections and launch pipeline.
def
__aiter__(self) -> AsyncIterator[str]:
1300 def __aiter__(self) -> AsyncIterator[str]: 1301 "Return asynchronous iterator over stdout/stderr." 1302 return self._readlines()
Return asynchronous iterator over stdout/stderr.
1304 @staticmethod 1305 async def run_pipeline(pipe: "shellous.Pipeline[Any]") -> str | Result: 1306 "Run a pipeline. This is the main entry point for PipeRunner." 1307 run = PipeRunner(pipe, capturing=False) 1308 async with run: 1309 pass 1310 1311 result = run.result() 1312 if pipe.options._return_result: 1313 return result 1314 return result.output
Run a pipeline. This is the main entry point for PipeRunner.
class
AuditEventInfo(typing.TypedDict):
75class AuditEventInfo(TypedDict): 76 """Info attached to each audit callback event. 77 78 See `audit_callback` in `Command.set` for more information. 79 """ 80 81 runner: Runner 82 "Reference to the Runner object." 83 84 failure: str 85 "When phase is 'stop', the name of the exception from starting the process." 86 87 signal: str 88 "When phase is 'signal', the signal name/number sent to the process."
Info attached to each audit callback event.
See audit_callback in Command.set for more information.