Command execution#

pglift.cmd.run(args: Sequence[str], *, input: str | None = None, check: bool = False, timeout: float | None = None, **kwargs: Any) → CompletedProcess#

Run a command as a subprocess while forwarning its stderr to module ‘logger’.

Standard output and errors of child subprocess are captured by default.

>>> run(["true"], input="a", capture_output=False)
CompletedProcess(args=['true'], returncode=0)

Files can also be used with stdout and stderr arguments:

>>> import tempfile
>>> with tempfile.NamedTemporaryFile() as f:
...     _ = run(["echo", "ahah"], stdout=f, stderr=None)
...     with open(f.name) as f:
...         print(f.read(), end="")
ahah

>>> r = run(["cat", "doesnotexist"], stdout=PIPE, stderr=PIPE, env={"LANG": "C"})
>>> print(r.stderr, end="")
cat: doesnotexist: No such file or directory

With check=True, CommandError is raised in case of non-zero return code:

>>> run(["cat", "doesnotexist"], check=True)
Traceback (most recent call last):
    ...
pglift.exceptions.CommandError: Command '['cat', 'doesnotexist']' returned non-zero exit status 1.

With a non-None timeout argument, a TimeoutExpired exception might be raised:

>>> run(["sleep", "0.1"], timeout=0.01)
Traceback (most recent call last):
    ...
subprocess.TimeoutExpired: Command '['sleep', '0.1']' timed out after 0.01 seconds

>>> run(["nosuchcommand", "x", "y", "-v"])
Traceback (most recent call last):
    ...
pglift.exceptions.FileNotFoundError: program from command 'nosuchcommand x y -v' not found
class pglift.cmd.Program(cmd: Sequence[str], pidfile: Path | None, *, timeout: float = 1, env: Mapping[str, str] | None = None, capture_output: bool = True)#

Start program described by ‘cmd’ and possibly store its PID in ‘pidfile’.

This is aimed at starting daemon programs.

Raises:

  • SystemError – if the program is already running.

  • CommandError – in case program execution terminates after timeout.

When used as a context manager, any exception raised within the block will trigger program termination at exit. This can be used to perform sanity checks shortly after program startup.

pglift.cmd.terminate_program(pidfile: Path) → None#

Terminate program matching PID in ‘pidfile’.

Upon successful termination, the ‘pidfile’ is removed. No-op if no process matching PID from ‘pidfile’ is running.

pglift.cmd.status_program(pidfile: Path) → Status#

Return the status of a program which PID is in ‘pidfile’.

Raises:

  • SystemError – if the program is already running.

  • CommandError – in case program execution terminates after timeout.