epycs 2.3.0

Epycs is a simple way to convert shell scripts to python

Epycs

Epycs is a simple way to convert shell scripts to python. It features

• A simple subprocess API
• A sane behaviour of exiting by default on subprocess failures
• A show-output-on-fail behaviour

The goal of this package is to be able to write shell-script equivalent code in python while still being terse, but adding a tons of goodness in terms of arithmetical expression, string manipulation, code reuse etc...

Say no to .sh and welcome .py with epycs, you'll thank me later.

Usage

This section is a complete, copy-pasteable cheat sheet of the public API. If you are an LLM/agent: everything you need to use epycs is below — no need to read the source or search the web.

pip install epycs   # or: uv add epycs

Run a command

cmd is a magic shell: any attribute is resolved to a program on PATH.

from epycs import cmd

cmd.ffmpeg("-i", "talk.mov", "talk.webm")  # transcode a video; progress streams to the terminal
cmd.git("clone", url, dest)                # args are str()-ed automatically (Path, int, ...)

By default a command prints to the terminal and returns a subprocess.CompletedProcess. To capture and parse the output instead, use out_filter (see below).

Default behaviours (the "goodness" over raw subprocess)

• Exit on failure: a non-zero return code exits your script with that code (like set -e). Disable globally with exit_on_error = False.
• Quiet-but-loud-on-fail: with quiet=True output is hidden, unless the command fails, in which case it is dumped to stderr.
• text=True is the default, so captured output is str, not bytes.

import epycs.subprocess as esp
esp.verbose = True          # print every command to stderr (like `set -x`)
esp.exit_on_error = False   # don't sys.exit() on non-zero; inspect .returncode

verbose and exit_on_error are also overridable without touching the globals — per call/per program with exit_on_error=, or for a block with the esp.options(...) context manager (handy for preformatted commands like ping/which that want a return code rather than a sys.exit):

rc = cmd.ping("-c1", host, exit_on_error=False).returncode   # this call only
ping = cmd.ping.arg(exit_on_error=False)                      # bake it into a program

with esp.options(exit_on_error=False, verbose=True):          # for the whole block
    ...

Capture & parse output: out_filter

out_filter captures stdout and runs it through a parser. Built-in filters (pass the name as a string) or any callable(str) -> object:

from epycs import cmd

sha     = cmd.git("rev-parse", "HEAD", out_filter="stripped")  # "9e7f6c2" (no trailing \n)
branches= cmd.git("branch", "--format=%(refname:short)", out_filter="text_lines")  # list[str]
files   = cmd.git("ls-files", "-z", out_filter="text_lines_0")  # list[str], NUL-split (space-safe)
config  = cmd.git("config", "--list", out_filter="key_value")   # {"user.email": "...", ...}
meta    = cmd.ffprobe("-show_format", "-print_format", "json", clip, out_filter="json")  # dict
running = cmd.docker("ps", "--format", "{{json .}}", out_filter="json_lines")  # list[dict]
users   = cmd.sqlite3("-csv", "-header", "app.db", "select * from users", out_filter="csv_dict")  # list[dict]
ver     = cmd.git("describe", "--tags", out_filter=lambda s: s.strip().removeprefix("v"))  # custom

Available built-in filters: text, stripped (whitespace-trimmed), text_lines, text_lines_0, key_value (Key=Value lines → dict), json, json_lines/ ndjson (one JSON value per line → list), xml, csv, csv_dict, and raw/bytes (binary passthrough, see below).

The same filters are importable as typed callables from epycs.filters, so you get completion and the call's return type is inferred (no more blanket Any). lines/lines_0 are the importable spellings of "text_lines" / "text_lines_0":

from epycs import cmd
from epycs.filters import json, lines, stripped

meta = cmd.ffprobe("-show_format", "-print_format", "json", clip, out_filter=json)  # parsed JSON
tags = cmd.git("tag", out_filter=lines)                   # inferred as list[str]
sha  = cmd.git("rev-parse", "HEAD", out_filter=stripped)  # inferred as str (no trailing \n)

Passing a string name still works and is now also typed: out_filter="text" infers str, out_filter="csv_dict" infers list[dict[str, str]], and a call with no out_filter returns the raw CompletedProcess.

Build & reuse commands: find_program, .arg(...) and [...]

find_program(name, *aliases) resolves a program once (trying aliases in order, raising ShellProgramNotFoundError if none found) and returns a reusable ShellProgram. .arg(...) returns a new command with extra args and/or default kwargs baked in (originals are never mutated):

from epycs import cmd, find_program

oci     = find_program("podman", "docker")               # first match wins (podman or docker)
ffprobe = cmd.ffprobe.arg("-v", "error", out_filter="stripped")  # quiet + capture baked in
ffprobe("-show_entries", "format=duration", "-of", "csv=p=0", clip)  # -> "12.480000"

git = cmd.git["-C", repo]                                # `git -C <repo> ...`
git("rev-parse", "--abbrev-ref", "HEAD", out_filter="stripped")  # -> current branch name

cmd.git.status(out_filter="text")  # attribute access also adds args -> `git status`

[...] is a readable alias for positional .arg(...), à la plumbum: cmd.git["-C", repo] is cmd.git.arg("-C", repo). Reach for .arg(...) when you also need to bake in keyword options (out_filter=, exit_on_error=, ...), which subscript syntax can't carry.

Predefined commands

A few everyday programs are exposed as ready-to-use attributes on cmd — no PATH lookup or alias dance needed:

from epycs import cmd

cmd.python("-m", "pip", "install", "-U", "pip")   # the *current* interpreter (sys.executable)
cmd.editor(path)                                   # $EDITOR (falls back to `vi`)
cmd.shell("-c", "ulimit -n")                       # $SHELL
for line in cmd.env():                             # `env`, pre-set to return list[str]
    key, _, value = line.partition("=")

cmd.python always points at the interpreter running your script — handy to re-invoke yourself or a stdlib module portably; cmd.env already carries out_filter="text_lines".

Keyword options

These epycs-specific kwargs work on any call; everything else is forwarded to subprocess.run (e.g. cwd=, timeout=, input=):

kwarg	effect
out_filter	capture stdout and parse it (see above)
quiet=True	hide output unless the command fails (then dump to stderr)
stdout_tee=True	capture stdout and still print it
binary=True	capture raw bytes (forces text=False); pair with the raw filter
exit_on_error=False	per-call override of the global exit-on-failure policy
background=True	start via Popen and return it immediately (no wait)
additional_env={"K": "v"}	add/override env vars; value None removes the var
additional_pathenv={"PATH": ["/x"]}	append entries to a PATH-like var

cmd.ffmpeg("-i", src, dst, quiet=True)           # silent unless it fails (then dumps the log)
proc = cmd.docker("logs", "-f", "web", background=True); proc.wait()  # Popen, run concurrently
cmd.make("-j8", cwd="build", additional_env={"CC": "clang"})

dl = cmd.curl["-fsSL", url].env(HTTPS_PROXY="http://proxy:3128")  # `.env(**vars)` == additional_env

Binary output

For media/archive commands the captured output is binary. Pass binary=True (it forces text=False) and use the raw filter (or any callable(bytes) -> object). This works on single commands and at the end of a pipe:

from epycs.filters import raw

png = cmd.magick("logo.svg", "png:-", out_filter=raw, binary=True)   # bytes
frame = (cmd.ffmpeg["-i", clip, "-vframes", "1", "png:-"]
         | cmd.magick["-", "-resize", "100x", "png:-"])(out_filter=raw, binary=True)

Detect whether a program is available

cmd.has(name, *aliases) is a soft check (no raise); cmd.require(name, *aliases, hint=...) resolves the program or raises a ShellProgramNotFoundError with an actionable install hint:

if not cmd.has("ffmpeg"):
    ...
ffmpeg = cmd.require("ffmpeg", hint="apt install ffmpeg")
# ShellProgramNotFoundError: 'ffmpeg' not found on PATH.
#   To install: apt install ffmpeg

which(name) returns the resolved executable as a contractme ExecutablePath (Path | None); it delegates to shutil.which, so it follows the same PATH rules as the rest of your tooling.

Pipe commands with |

Chain commands with | exactly like a shell pipe. Operands are uncalled programs (bake args with [...] or .arg(...)); the pipeline runs only when you call it:

from epycs import cmd

# stdout of each stage is wired to stdin of the next: 1 fps of PNG frames, tiled into a strip
(cmd.ffmpeg["-i", clip, "-vf", "fps=1", "-f", "image2pipe", "-vcodec", "png", "-"]
 | cmd.magick["-", "-resize", "240x", "+append", "strip.png"])()

# args/kwargs of the terminal call go to the last stage
errs = (cmd.journalctl["-u", "nginx", "-o", "cat"] | cmd.grep)("error", out_filter="text_lines")

Pipelines behave like a shell pipe with set -o pipefail: the pipeline's return code is the rightmost non-zero stage code (0 if all succeed). Combined with the default exit_on_error, any failing stage aborts the script — so a failing curl in curl | jq aborts instead of being silently masked by jq's 0 (a classic bash footgun).

import epycs.subprocess as esp
esp.exit_on_error = False
(cmd.curl["-fsS", url] | cmd.jq["."])().returncode   # -> curl's nonzero, not jq's 0

Extras

from epycs.subprocess import source_shell_script, python_to_subprocess
from epycs.config import load_from

source_shell_script("env.sh")   # apply a shell script's env to os.environ

# load_from looks at $NAME_CONFIG then ~/.config/<name>/<name>.{toml,json,...}
cfg = load_from("myapp")        # returns parsed config, or None if absent

@python_to_subprocess            # turn a python function into a pipeable subprocess
def worker(cmd_open): ...

Advanced recipes

Run in the background & concurrently

background=True starts the command via Popen and returns it immediately, so you can launch several and synchronise only when you need the result:

from epycs import cmd

build = cmd.docker("build", "-t", "app:ci", ".", background=True)   # returns a Popen now
encode = cmd.ffmpeg("-i", master, "-c:v", "libx264", out, background=True)

build.wait()                       # block only when you actually need it
assert build.returncode == 0
encode.terminate()                 # ... and stop the other one

(Pipelines already run their stages concurrently — background is for whole commands you want to supervise yourself.)

Write your own filter

An out_filter is just any callable(str) -> object (or callable(bytes) -> object with binary=True), so a function or lambda is enough:

def repo_tags(s: str) -> list[str]:
    return [line for line in s.splitlines() if line and "<none>" not in line]

tags = cmd.docker("images", "--format", "{{.Repository}}:{{.Tag}}", out_filter=repo_tags)

Give it the same runtime contracts as the built-ins with @contractme.annotated, and register a string name in epycs.filters.BY_NAME for out_filter="…" access:

from contractme import annotated
from contractme.types import JsonStr
from epycs.filters import BY_NAME

@annotated
def json_keys(s: JsonStr) -> list:          # input validated as well-formed JSON
    import json
    return list(json.loads(s))

BY_NAME["json_keys"] = json_keys
fields = cmd.kubectl("get", "pod", "web", "-o", "json", out_filter="json_keys")

Run inside an environment (cmd.source / cmd.activate)

epycs resolves a program on PATH at call time, so the generic way to "run inside X" is to pull X's environment into your process and keep going. cmd.source(path) does exactly that — the "activate this" button for any shell environment: a profile script, a vendor env.sh, a cross-compilation toolchain, an SDK setup, …

Because source is a shell builtin (there is no source binary), the name can never collide with a real program, so epycs repurposes it — something even plumbum doesn't offer.

from epycs import cmd

cmd.source("/opt/intel/oneapi/setvars.sh")   # a vendor SDK env.sh
cmd.icx("-O3", "kernel.c", "-o", "kernel")    # now on PATH

How it works — and why it matters. cmd.source picks the capture mechanism from the file's dialect: a .ps1 is dot-sourced by PowerShell, a .bat/.cmd by cmd.exe, an activate_this.py is run in-process, and anything else is sourced by your $SHELL ($SHELL -c 'source <path> && env -0'). So a plain shell script must be written in that shell's dialect.

That bites with virtualenvs, which ship one activation script per shell — activate (POSIX/bash/zsh), activate.fish, activate.csh, Activate.ps1 (PowerShell), activate.bat (cmd.exe), plus the shell-independent activate_this.py — none interchangeable. You usually don't know (or want to hard-code) the user's shell, so don't pick the file yourself:

cmd.activate(".venv")     # the right script for the shell/OS — you don't choose
cmd.pytest("-q")          # now the venv's pytest, whatever shell you're in

cmd.activate prefers activate_this.py when the venv ships one (virtualenv does; the stdlib venv does not) — it's exec'd in-process, so it works on any OS/shell and also makes the venv importable. Otherwise it falls back to venv_activation_script(venv, shell=None), the helper that returns the script matching $SHELL (or Activate.ps1 on Windows) — usable standalone too.

Caveats of the $SHELL path (not the .py one): $SHELL must be set, the script must match its dialect (a bash env.sh won't source under fish), and it relies on GNU env -0 (Linux/coreutils).

For a uv-managed project you don't need to activate anything — let uv select the environment per call:

cmd.uv("run", "pytest", "-q")

Advanced wrappers (epycs.wrappers)

Thin, simplified wrappers around complex tools. Each is a thin layer over the DSL above: the tool is exposed as a lazily-resolved ShellProgram (program()) plus helpers that bake in flags and parse output through an out_filter — so you get list[dict] / str instead of a raw CompletedProcess. Importing a wrapper never fails when the tool is missing; the actionable "not found / install X" error is raised only on first use. Domain inputs are validated by contractme contracts (epycs.types).

from epycs.wrappers import docker, systemctl, journalctl, ffmpeg, magick

for c in docker.ps(all_=True):            # list[dict] (parsed `docker ps`)
    print(c["Names"], c["Status"])
cid = docker.run("nginx", detach=True)    # -> container id (str)

if not systemctl.is_active("nginx"):       # bool
    systemctl.restart("nginx")            # or restart("x", user=True) for --user
pid = systemctl.show("nginx")["MainPID"]  # `systemctl show` as a dict

for e in journalctl.entries(unit="nginx", since="today", priority="err"):
    print(e["MESSAGE"])                   # list[dict] via `-o json`

ffmpeg.convert("clip.mov", "clip.mp4", "-c:v", "libx264")  # -y by default
secs = ffmpeg.duration("clip.mp4")        # float (via ffprobe -print_format json)
magick.resize("a.png", "thumb.png", "200x200")
meta = magick.info("a.png")               # list[dict] via the `json:` coder

Remote commands: epycs.wrappers.ssh (extra)

SSH-as-a-shell is delegated to Fabric — install the extra: pip install 'epycs[ssh]' (or uv add 'epycs[ssh]'). Fabric is imported lazily. A non-zero remote exit raises SshCommandError naming the host, command, exit code and stderr tail; with epycs.subprocess.verbose each hop logs + [host] command. The remote env is not inherited — pass it with env=.

from epycs.wrappers import ssh

r = ssh.run("user@host", "uptime")                 # fabric Result; raises on failure
print(r.stdout)

# recursive `ssh dest cmd`: prefer an explicit jump host over nesting ssh in the
# command string, so the failing host is unambiguous
web = ssh.connect("web1", gateway=ssh.connect("bastion"))
ssh.run(web, "systemctl is-active nginx", env={"LC_ALL": "C"}, cwd="/srv")

Changelog

See CHANGELOG.md for the full, versioned history.