shellous 0.2.0

shellous: Run Processes and Pipelines

shellous: Run Processes and Pipelines

shellous is an asyncio library that provides a concise API for running subprocesses. It is similar to and inspired by sh.

import asyncio
import shellous

sh = shellous.context()

async def main():
    result = await (sh("ls") | sh("grep", "README"))
    print(result)

asyncio.run(main())

Benefits

• Run programs asychronously in a single line.
• Easily capture output or redirect stdin, stdout and stderr to files.
• Easily construct pipelines.
• Runs on Linux, MacOS and Windows.

Requirements

• Requires Python 3.9 or later.
• Requires an asyncio event loop.

Basic Usage

Start the asyncio REPL by typing python3 -m asyncio, and import the shellous module:

>>> import shellous

Before we can do anything else, we need to create a context. Store the context in a short variable name like sh because we'll be typing it a lot.

>>> sh = shellous.context()

Now, we're ready to run our first command. Here's one that runs echo "hello, world".

>>> await sh("echo", "hello, world")
'hello, world\n'

The first argument is the program name. It is followed by zero or more separate arguments.

A command does not run until you await it. Here, we create our own echo command with "-n" to omit the newline. Note, echo("abc") is the same as echo -n "abc".

>>> echo = sh("echo", "-n")
>>> await echo("abc")
'abc'

More on commands...

Results and Exit Codes

When you await a command, it captures the standard output and returns it. You can optionally have the command return a Result object. The Result object will contain more information about the command execution including the exit_code. To return a result object, set return_result option to True.

>>> await echo("abc").set(return_result=True)
Result(output_bytes=b'abc', exit_code=0, cancelled=False, encoding='utf-8', extra=None)

The above command had an exit_code of 0.

If a command exits with a non-zero exit code, it raises a ResultError exception that contains the Result object.

>>> await sh("cat", "does_not_exist")
Traceback (most recent call last):
  ...
shellous.result.ResultError: Result(output_bytes=b'', exit_code=1, cancelled=False, encoding='utf-8', extra=None)

More on results...

Redirecting Standard Input

You can change the standard input of a command by using the | operator.

>>> cmd = "abc" | sh("wc", "-c")
>>> await cmd
'       3\n'

To redirect stdin using a file's contents, use a Path object from pathlib.

>>> from pathlib import Path
>>> cmd = Path("README.md") | sh("wc", "-l")
>>> await cmd
'     210\n'

More on redirection...

Redirecting Standard Output

To redirect standard output, use the | operator.

>>> 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'

More on redirection...

Redirecting Standard Error

By default, standard error is not captured. To redirect standard error, use the stderr method.

>>> cmd = sh("cat", "does_not_exist").stderr(shellous.STDOUT)
>>> await cmd.set(exit_codes={0,1})
'cat: does_not_exist: No such file or directory\n'

You can redirect standard error to a file or path.

To redirect standard error to the hosting program's sys.stderr, use the INHERIT redirect option.

>>> cmd = sh("cat", "does_not_exist").stderr(shellous.INHERIT)
>>> await cmd
cat: does_not_exist: No such file or directory
Traceback (most recent call last):
  ...
shellous.result.ResultError: Result(output_bytes=b'', exit_code=1, cancelled=False, encoding='utf-8', extra=None)

More on redirection...

Pipelines

You can create a pipeline by combining commands using the | operator.

>>> pipe = sh("ls") | sh("grep", "README")
>>> await pipe
'README.md\n'

Iteration

You can loop over a command's output.

>>> async for line in pipe:
...   print(line.rstrip())
... 
README.md

Async With

You can use async with to interact with the process streams directly. You have to be careful; you are responsible for correctly reading and writing multiple streams at the same time.

>>> runner = pipe.runner()
>>> async with runner as (stdin, stdout, stderr):
...   data = await stdout.readline()
...   print(data)
... 
b'README.md\n'