Run a command

Declare, build, run

A command starts as a program in a catalogue. sh.make consults the catalogue and returns a builder; calling the builder makes a SafeCmd, an exact argument vector for that program; running it returns a CommandResult. The default catalogue holds common tools such as ECHO, GIT, and TAR. For an application's own executable, build a catalogue that names it:

import sys

from cuprum import Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="my-script")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python("-c", "print('hello, cuprum!')")
result = command.run_sync()
assert result.ok
assert result.stdout == "hello, cuprum!\n"
assert result.stderr == ""

run_sync() suits a synchronous script; in an async application, await command.run() inside a coroutine. Both return the same result. The argument vector goes to the child without a shell, so quoting, globs, and pipes inside a string are passed through literally rather than interpreted.

Three ways a run can fail

Each failure happens at a different moment, and each means something different:

• UnknownProgramError when the builder is made. The program is not in the catalogue: a configuration or policy mistake, raised before anything runs.
• FileNotFoundError at launch. The program is catalogued, but the operating system cannot find it: check the path, or the child's inherited PATH.
• A result with ok == False. The child ran and exited non-zero. Cuprum does not raise; read exit_code and decide by the tool's own exit-code contract.

import sys

from cuprum import Program, ProgramCatalogue, UnknownProgramError, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="failures")
try:
    sh.make(Program("not-in-this-catalogue"), catalogue=catalogue)
except UnknownProgramError:
    pass
else:
    raise AssertionError("an unknown program should be rejected")

python = sh.make(Program(sys.executable), catalogue=catalogue)
result = python("-c", "import sys; sys.exit(7)").run_sync()
assert not result.ok
assert result.exit_code == 7

Read stderr for the tool's diagnostics, but never parse it to decide whether Cuprum's catalogue accepted the program: that is what the exception is for.

Build arguments deliberately

The builder accepts exactly ArgValue: strings, numbers, booleans, and paths, as positional or keyword values. Anything else, a list, a dict, or bytes, raises TypeError naming the type rather than reaching the child as its repr; decode filesystem bytes with os.fsdecode() first. Keyword arguments become --name=value, with underscores turned into hyphens, which suits only tools that accept that form: for a bare flag such as --check, pass it positionally, since check=True would produce --check=True. None raises TypeError in either position, so decide before building whether an optional flag is omitted or replaced. An argument containing spaces stays one argument.

import sys

from cuprum import Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="arguments")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python("-c", "print('two words')")
assert command.argv == ("-c", "print('two words')")
assert command.argv_with_program == (sys.executable, *command.argv)

Names that configure how a command runs are reserved: every ExecutionContext field (env, env_mode, cwd, cancel_grace, native_pump_cleanup_grace, timeout, stdout_sink, stderr_sink, encoding, errors, tags, and process_group) and the parameters of run_sync (output, timeout, context, and stdin). Passing one to the builder raises TypeError naming the right place for it, instead of quietly rendering --cwd=<path> for a tool that might ignore it and run in the wrong directory:

import sys

from cuprum import Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="reserved")
python = sh.make(Program(sys.executable), catalogue=catalogue)
try:
    python("-c", "print('unused')", cwd="/tmp")
except TypeError as exc:
    assert str(exc) == (
        "cwd is an execution option; pass ExecutionContext(cwd=...) to run_sync"
    )
else:
    raise AssertionError("cwd should not become a child flag")
try:
    python("-c", "print('unused')", timeout=5)
except TypeError as exc:
    assert str(exc) == "timeout is an execution option; pass timeout=... to run_sync"
else:
    raise AssertionError("timeout should not become a child flag")

A tool whose command line genuinely takes such a flag still receives it positionally, as "--cwd=<dir>", and names merely resembling the reserved ones, such as working_dir, are unaffected. For Git, rsync, and tar, the typed builders in cuprum.builders validate paths, refs, and options when the command is built; see Catalogues and builders.

Hold a running child and decide when to stop it

run() owns its child from start to finish. When the caller must decide when the child stops, SafeCmd.start spawns it and hands back an owned RunningCommand instead of awaiting it. Await the handle's wait() for the outcome, read result once it has settled, or stop the child with terminate(), which sends SIGTERM, waits the cancel grace, then SIGKILL, or kill(), which sends SIGKILL without asking. The async with block is what guarantees teardown: leaving it stops a child still running.

import asyncio
import sys

from cuprum import ExecutionContext, Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="handle")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python("-c", "import time; time.sleep(30)")


async def stop_a_child() -> int | None:
    async with await command.start(
        context=ExecutionContext(cancel_grace=0.2),
    ) as handle:
        assert handle.pid is not None and handle.result is None
        result = await handle.terminate()
        assert result.exit_code == -15
        return handle.result.exit_code if handle.result is not None else None


assert asyncio.run(stop_a_child()) == -15

A spawn that cannot start belongs to start() itself: it raises the spawn's own failure and returns no handle, so there is nothing to clean up. wait(timeout=...) bounds one call; its expiry stops the child the way a timed-out run() does, raises TimeoutExpired with what the child had written, and records the outcome, so a later wait() or terminate() on the same handle reports the exit code rather than an error. Capture is always on for a handle, so a stopped child still reports its output. The example's -15 is the POSIX convention for a child ended by SIGTERM; one that had to be killed reports -9.

signal() delivers one POSIX signal without settling the run, and has two refusals because the remedy differs: SignalAfterExitError, a ProcessLookupError, means the child has gone, and UnsupportedSignalError means the platform has no route to a caller-named signal, so terminate() or kill() is the only option there. Stopping an already-exited child is not an error: both are no-ops that report the recorded outcome, so they are safe to call twice or from two places at once. What each control operation did to the child can be observed; see Observe production runs.

Migrate a subprocess call

Moving a subprocess call to Cuprum is mostly a matter of moving each argument to its new home. Here is the same command before and after:

import subprocess
import sys
import tempfile
from pathlib import Path

from cuprum import ExecutionContext, Program, ProgramCatalogue, sh

script = "import os; print(os.path.basename(os.getcwd()))"
workdir = tempfile.mkdtemp()

# Before: subprocess, checking the exit status and capturing text.
before = subprocess.run(
    [sys.executable, "-c", script],
    capture_output=True,
    text=True,
    check=True,
    cwd=workdir,
)

# After: declare the program once, then build and run the command.
catalogue = ProgramCatalogue.from_programs(sys.executable, name="migrated")
python = sh.make(Program(sys.executable), catalogue=catalogue)
result = python("-c", script).run_sync(context=ExecutionContext(cwd=workdir))
if not result.ok:
    raise RuntimeError(f"exited with {result.exit_code}: {result.stderr}")

assert result.stdout == before.stdout == f"{Path(workdir).name}\n"
Path(workdir).rmdir()

Note the environment row: subprocess replaces the child's environment with env=, whereas Cuprum layers ExecutionContext.env over the parent's unless EnvMode.REPLACE asks for a replacement; see Input, environment, and deadlines. An application upgrading from Cuprum 0.1 should read Migrate to 0.2.0 instead.