duct.py

Duct is a library for running child processes. Duct makes it easy to build pipelines and redirect IO like a shell. At the same time, Duct helps you write correct, portable code: whitespace is never significant, errors from child processes get reported by default, and a variety of gotchas, bugs, and platform inconsistencies are handled for you the Right Way™.

• Documentation
• PyPI package
• the same library, in Rust

Changelog

• v0.6.3
  • Added Handle.pids and ReaderHandle.pids.
• v0.6.2
  • Added ReaderHandle.try_wait.
• v0.6.1
  • Added ReaderHandle.kill.
  • Kill methods no longer wait on IO threads to complete. This avoids blocking on unkilled grandchildren.
• v0.6.0
  • The kill method now reaps killed child processes before returning.
  • Removed the sh function.
  • Removed the then method.
  • Added Handle.kill.
  • Added ReaderHandle and Expression.reader.
  • Added Expression.stdout_stderr_swap.
  • Added Expression.before_spawn.
  • Renamed stdin/stdout/stderr to stdin_path/stdout_path/stderr_path.
  • Renamed input to stdin_bytes.
  • This will be the last major release supporting Python 2.

Examples

Run a command without capturing any output. Here "hi" is printed directly to the terminal:

>>> from duct import cmd
>>> cmd("echo", "hi").run()
hi
Output(status=0, stdout=None, stderr=None)

Capture the standard output of a command. Here "hi" is returned as a string:

>>> cmd("echo", "hi").read()
'hi'

Capture the standard output of a pipeline:

>>> cmd("echo", "hi").pipe(cmd("sed", "s/i/o/")).read()
'ho'

Merge standard error into standard output and read both incrementally:

>>> big_cmd = cmd("bash", "-c", "echo out && echo err 1>&2")
>>> reader = big_cmd.stderr_to_stdout().reader()
>>> with reader:
...     reader.readlines()
[b'out\n', b'err\n']

Children that exit with a non-zero status raise an exception by default:

>>> cmd("false").run()
Traceback (most recent call last):
...
duct.StatusError: Expression cmd('false') returned non-zero exit status: Output(status=1, stdout=None, stderr=None)
>>> cmd("false").unchecked().run()
Output(status=1, stdout=None, stderr=None)