Metadata-Version: 2.4
Name: epycs
Version: 2.3.0
Summary: Epycs is a simple way to convert shell scripts to python
Requires-Python: >=3.13
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: contractme>=1.8.0
Requires-Dist: pytest>=9.0.1
Provides-Extra: ssh
Requires-Dist: fabric>=3; extra == "ssh"
Provides-Extra: all
Requires-Dist: fabric>=3; extra == "all"
Dynamic: license-file

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

```sh
pip install epycs   # or: uv add epycs
```

## Run a command

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

```python
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`.

```python
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`):

```python
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`:

```python
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"`:

```python
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):

```python
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:

```python
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 |

```python
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:

```python
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:

```python
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`](https://pypi.org/project/contractme/) (`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:

```python
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).

```python
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

```python
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:

```python
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:

```python
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:

```python
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.

```python
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:

```python
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:

```python
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`).

```python
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](https://www.fabfile.org/) — 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=`.

```python
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](CHANGELOG.md) for the full, versioned history.
