Skip to main content

shellous: Run Processes and Pipelines

Project description

shellous: Run Processes and Pipelines

CI codecov

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("echo", "hello, world")
    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.

Requirements

  • Requires Python 3.9 or later.
  • Runs on Linux, MacOS and Windows.

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. This is our chance to customize settings like environment variables or text encoding. 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'

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, ...

To return a Result object instead of raising an exception, set the allowed_exit_codes option together with the return_result option.

>>> await sh("cat", "does_not_exist").set(return_result=True, allowed_exit_codes={0,1})
Result(output_bytes=b'', exit_code=1, cancelled=False, encoding='utf-8', extra=None)

You can wire this into a cat command to do it automatically.

>>> cat = sh("cat").set(return_result=True, allowed_exit_codes={0,1})
>>> await cat("does_not_exist")
Result(output_bytes=b'', exit_code=1, cancelled=False, encoding='utf-8', extra=None)

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("/usr/bin/wc") | sh("wc", "-c")
>>> await cmd
'  137968\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(allowed_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, ...

More on redirection...

Pipelines

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

>>> pipe = sh("echo", "abc") | sh("wc", "-c")
>>> await pipe
'       4\n'

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

shellous-0.1.0.tar.gz (18.0 kB view hashes)

Uploaded Source

Built Distribution

shellous-0.1.0-py3-none-any.whl (18.4 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page