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