A shell for subprocess
Project description
subprocess_shell
is a Python package providing an alternative interface to sub processes. The aim is simplicity comparable to shell scripting and transparency for more complex use cases.
[[TOC]]
Features
- Simple
- e.g. 4 functions (
start
,write
,wait
,read
) and 3 operators (>>
,+
,-
)
- e.g. 4 functions (
- Transparent
- usability layer for subprocess except streams
- Separates streams
- no interleaving of stdout and stderr and from different processes of a chain
- Avoids deadlocks due to OS pipe buffer limits by using queues
- Uses Rich if available
images/rich_output.png
images/rich_output.png
Examples
|
subprocess_shell |
subprocess |
Plumbum[^e1] |
|
---|---|---|---|---|
initialization |
from subprocess_shell import *
|
import subprocess
|
from plumbum import local
|
|
run command |
echo this
|
["echo", "this"] >> start() >> wait()
|
assert subprocess.Popen(["echo", "this"]).wait() == 0
|
local["echo"]["this"].run_fg()
|
redirect stream |
echo this > /path/to/file
|
["echo", "this"] >> start(stdout="/path/to/file") >> wait()
|
with open("/path/to/file", "wb") as stdout:
assert subprocess.Popen(["echo", "this"], stdout=stdout).wait() == 0
|
(local["echo"]["this"] > "/path/to/file").run_fg()
|
read stream |
a=$(echo this)
|
a = ["echo", "this"] >> start() >> read()
|
process = subprocess.Popen(["echo", "this"], stdout=subprocess.PIPE)
a, _ = process.communicate()
assert process.wait() == 0
|
a = local["echo"]("this")
|
write stream |
cat - <<EOF
this
EOF
|
["cat", "-"] >> start() >> write("this") >> wait()
|
process = subprocess.Popen(["cat", "-"], stdin=subprocess.PIPE)
process.communicate(b"this")
assert process.wait() == 0
|
(local["cat"]["-"] << "this").run_fg()
|
chain commands |
echo this | cat -
|
["echo", "this"] >> start() + ["cat", "-"] >> start() >> wait()
|
process = subprocess.Popen(["echo", "this"], stdout=subprocess.PIPE)
assert subprocess.Popen(["cat", "-"], stdin=process.stdout).wait() == 0
assert process.wait() == 0
|
(local["echo"]["this"] | local["cat"]["-"]).run_fg()
|
branch out | ? |
import sys
arguments = [sys.executable, "-c", "import sys; print('stdout'); print('stderr', file=sys.stderr)"]
process = arguments >> start(pass_stdout=True, pass_stderr=True)
process + ["cat", "-"] >> start() >> wait()
process - ["cat", "-"] >> start() >> wait()
|
import sys
arguments = [sys.executable, "-c", "import sys; print('stdout'); print('stderr', file=sys.stderr)"]
process = subprocess.Popen(arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
assert subprocess.Popen(["cat", "-"], stdin=process.stdout).wait() == 0
assert subprocess.Popen(["cat", "-"], stdin=process.stderr).wait() == 0
assert process.wait() == 0
|
not supported[^e2] |
errors in chains | ? |
["echo", "this"] >> start(return_codes=(0, 1)) - ["cat", "-"] >> start() >> wait(return_codes=(0, 2))
|
first_process = subprocess.Popen(["echo", "this"], stderr=subprocess.PIPE)
second_process = subprocess.Popen(["cat", "-"], stdin=first_process.stderr)
assert first_process.wait() in (0, 1) and second_process.wait() in (0, 2)
|
not supported[^e2] |
callbacks |
["echo", "this"] >> start(stdout=print) >> wait()
|
process = subprocess.Popen(["echo", "this"], stdout=subprocess.PIPE)
for bytes in process.stdout:
print(bytes)
assert process.wait() == 0
!![^e3] |
[^e1]: Mostly adapted versions from https://www.reddit.com/r/Python/comments/16byt8j/comment/jzhh21f/?utm_source=share&utm_medium=web2x&context=3 [^e2]: Has been requested years ago [^e3]: This is very limited and has several issues with potential for deadlocks. An exact equivalent would be too long for this table.
Notes
bash -e
because errors can have serious consequences- e.g.
a=$(failing command)
sudo chown -R root:root "$a/"
assert process.wait() == 0
is the shortest (readable) code waiting for a process to stop and asserting the return code- complexity of code for Plumbum can be misleading because it has a much wider scope (e.g. remote execution and files)
Quickstart
-
Prepare virtual environment (optional but recommended)
- e.g. Pipenv:
python -m pip install -U pipenv
- e.g. Pipenv:
-
Install subprocess_shell
- e.g.
python -m pipenv run pip install subprocess_shell
- e.g.
-
Import and use it
- e.g.
from subprocess_shell import *
andpython -m pipenv run python ...
- e.g.
-
Prepare tests
- e.g.
python -m pipenv run pip install subprocess_shell[test]
- e.g.
-
Run tests
- e.g.
python -m pipenv run pytest ./tests
- e.g.
Documentation
from subprocess_shell import *
Start process
process = arguments >> start(
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
pass_stdout=False,
stderr=subprocess.PIPE,
pass_stderr=False,
queue_size=0,
return_codes=(0,),
**{},
)
|
iterable |
arguments are converted to string using |
|
|
provide stdin |
any |
same as |
|
|
|
provide stdout |
string or |
redirect stdout to file |
|
|
call function for each chunk from stdout | |
any |
same as |
|
|
|
if |
|
don't use stdout | |
|
|
provide stderr |
string or |
redirect stderr to file |
|
|
call function for each chunk from stderr | |
any |
same as |
|
|
|
if |
|
don't use stderr | |
|
|
no limit on size of queues |
|
wait for other threads to process queues if full; !! can lead to deadlocks !! |
|
|
|
if in a chain: analog of |
collection |
if in a chain: analog of |
|
|
if in a chain: analog of |
|
|
|
passed to |
Write to stdin
process = process >> write(argument)
|
string or |
en/decoded if necessary, written to stdin and flushed |
requires start(stdin=subprocess.PIPE)
Wait for process
return_code = process >> wait(
stdout=True,
stderr=True,
return_codes=(0,),
)
|
|
if stdout is queued: collect stdout, format and print to stdout |
|
don't use stdout | |
any |
if stdout is queued: collect stdout, format and print with |
|
|
|
if stderr is queued: collect stderr, format and print to stderr |
|
don't use stderr | |
any |
if stderr is queued: collect stderr, format and print with |
|
|
|
assert the return code is 0 |
collection | assert the return code is in the collection | |
|
don't assert the return code |
Read from stdout/stderr
string = process >> read(
stdout=True,
stderr=False,
bytes=False,
return_codes=(0,),
)
|
|
execute |
|
execute |
|
any |
execute |
|
|
|
execute |
|
execute |
|
any |
execute |
|
|
|
return a string or tuple of strings |
|
return |
|
|
|
execute |
any |
execute |
process.get_stdout_lines(bytes=False)
process.get_stderr_lines(bytes=False)
process.get_stdout_strings()
process.get_stderr_strings()
process.get_stdout_bytes()
process.get_stderr_bytes()
process.get_stdout_objects()
process.get_stderr_objects()
|
|
return iterable of strings |
|
return iterable of |
requires queued stdout/stderr
Chain processes / pass streams
process = source_arguments >> start(...) + arguments >> start(...)
# or
source_process = source_arguments >> start(..., pass_stdout=True)
process = source_process + arguments >> start(...)
process = source_arguments >> start(...) - arguments >> start(...)
# or
source_process = source_arguments >> start(..., pass_stderr=True)
process = source_process - arguments >> start(...)
source_process = process.get_source_process()
process >> wait(...)
waits for the processes from left/source to right/target
Limitations
- Linux only
Motivation
Shell scripting is great for simple tasks. When tasks become more complex, e.g. hard to chain or require non-trivial processing, I always switch to Python. The interface provided by subprocess is rather verbose and parts that would look trivial in a shell script end up a repetitive mess. After refactoring up the mess once too often, it was time for a change.
See also
Why the name subprocess_shell
Simply because I like the picture of subprocess with a sturdy layer that is easy and safe to handle.
Also, while writing import subprocess
it is easy to remember to add _shell
.
Before subprocess_shell I chose to call it shell. This was a bad name for several reasons, but most notably because the term shell is commonly used for applications providing an interface to the operating system.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.