Command execution

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

pglift.system.cmd.start_program(cmd: ~collections.abc.Sequence[str], *, pidfile: ~pathlib.Path | None, logfile: ~pathlib.Path | None, timeout: float = 1, env: ~collections.abc.Mapping[str, str] | None = None, fs: ~typing.Annotated[~pglift.system.fs.AbstractFS[~pathlib.PurePath] | ~pglift.system.fs.AbstractFS[~pathlib.Path], ~pglift.deps.Dependency(var=<ContextVar name='FileSystem' default=<pglift.system.fs.LocalFS object at 0x76365c6efb60> at 0x76365c60a520>)] = <object object>) → Popen[bytes]

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

Raises:

• SystemError – if the program is already running.

• CommandError – in case program unexpectedly terminates before timeout.

pglift.system.cmd.terminate_program(pidfile: ~pathlib.Path, *, fs: ~typing.Annotated[~pglift.system.fs.AbstractFS[~pathlib.PurePath] | ~pglift.system.fs.AbstractFS[~pathlib.Path], ~pglift.deps.Dependency(var=<ContextVar name='FileSystem' default=<pglift.system.fs.LocalFS object at 0x76365c6efb60> at 0x76365c60a520>)] = <object object>) → 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.system.cmd.status_program(pidfile: ~pathlib.Path, *, fs: ~typing.Annotated[~pglift.system.fs.AbstractFS[~pathlib.PurePath] | ~pglift.system.fs.AbstractFS[~pathlib.Path], ~pglift.deps.Dependency(var=<ContextVar name='FileSystem' default=<pglift.system.fs.LocalFS object at 0x76365c6efb60> at 0x76365c60a520>)] = <object object>) → 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.