os.process

Suprocesses management.

This module provides some functions and classes to ease spawn of processes in blocking or non blocking mode, redirection of its stdout, stderr and stdin. It also provides some helpers to check the process status

Module Contents

Classes

Run	Class to handle processes.
File	Can be a PIPE, a file object.

Functions

subprocess_setup(→ None)	Reset SIGPIPE handler.
get_rlimit(→ str)
quote_arg(→ str)	Return the quoted version of the given argument.
to_cmd_lines(→ list[CmdLine])
command_line_image(→ str)	Return a string image of the given command(s).
enable_commands_handler(→ logging.Handler)	Add a handler that log all commands launched with Run in a file.
disable_commands_handler(→ None)	Disable special handler for commands.
wait_for_processes(→ int | None)	Wait for several processes spawned with Run.
is_running(→ bool)	Check whether a process with the given pid is running.
kill_process_tree(→ bool)	Kill a hierarchy of processes.

Attributes

CmdLine
logger
CMD_LOGGER_NAME
cmdlogger
has_psutil

os.process.CmdLine

os.process.logger

os.process.CMD_LOGGER_NAME = 'os.process.cmdline'

os.process.cmdlogger

os.process.has_psutil = True

os.process.subprocess_setup() → None

Reset SIGPIPE handler.

Python installs a SIGPIPE handler by default. This is usually not what non-Python subprocesses expect.

os.process.get_rlimit(platform: str | None = None) → str

os.process.quote_arg(arg: str) → str

Return the quoted version of the given argument.

Returns a human-friendly representation of the given argument, but with all extra quoting done if necessary. The intent is to produce an argument image that can be copy/pasted on a POSIX shell command (at a shell prompt). :param arg: argument to quote

os.process.to_cmd_lines(cmds: AnyCmdLine) → list[CmdLine]

os.process.command_line_image(cmds: AnyCmdLine) → str

Return a string image of the given command(s).

Parameters:

cmds – Same as the cmds parameter in the Run.__init__ method.

This method also handles quoting as defined for POSIX shells. This means that arguments containing special characters (such as a simple space, or a backslash, for instance), are properly quoted. This makes it possible to execute the same command by copy/pasting the image in a shell prompt.

The result is expected to be a string that can be sent verbatim to a shell for execution.

os.process.enable_commands_handler(filename: str, mode: str = 'a') → logging.Handler

Add a handler that log all commands launched with Run in a file.

Parameters:

• filename – path to log the commands

• mode – mode used to open the file (default is ‘a’)

Returns:

the added handler

os.process.disable_commands_handler(handler: logging.Handler) → None

Disable special handler for commands.

Parameters:

handler – Handler returned by enable_commands_handler

class os.process.Run(cmds: AnyCmdLine, cwd: str | None = None, output: DEVNULL_VALUE | PIPE_VALUE | str | IO | None = PIPE, error: STDOUT_VALUE | DEVNULL_VALUE | PIPE_VALUE | str | IO | None = STDOUT, input: DEVNULL_VALUE | PIPE_VALUE | str | bytes | IO | None = None, bg: bool = False, timeout: int | None = None, env: dict | None = None, set_sigpipe: bool = True, parse_shebang: bool = False, ignore_environ: bool = True)

Class to handle processes.

Variables:

• cmds – The cmds argument passed to the __init__ method (a command line passed in a list, or a list of command lines passed as a list of list).

• status –

  The exit status. As the exit status is only meaningful after the process has exited, its initial value is None (see __init__’s “bg” parameter for more information about this).

  When a problem running the command is detected and a process does not get created, its value gets set to the special value 127.

• raw_out – process standard output as bytes (if instanciated with output = PIPE). Use self.out to get a decoded string.

• raw_err – same as raw_out but for standard error.

• pid – PID. Set to -1 if the command failed to run.

property out: str | None

Process output as string.

Attempt is done to decode as utf-8 the output. If the output is not in utf-8 a string representation will be returned (see e3.text.bytes_as_str).

property err: str | None

Process error as string.

Attempt is done to decode as utf-8 the output. If the output is not in utf-8 a string representation will be returned (see e3.text.bytes_as_str).

command_line_image() → str

Get shell command line image of the spawned command(s).

This just a convenient wrapper around the function of the same name.

close_files() → None

Close all file descriptors.

__error(error: Exception, cmds: list[CmdLine]) → None

Set pid to -1 and status to 127 before closing files.

wait() → int

Wait until process ends and return its status.

Returns:

exit code of the process

poll() → int | None

Check the process status and set self.status if available.

This method checks whether the underlying process has exited or not. If it hasn’t, then it just returns None immediately. Otherwise, it stores the process’ exit code in self.status and then returns it.

Returns:

None if the process is still alive; otherwise, returns the process exit status.

kill(recursive: bool = True, timeout: int = 3) → None

Kill the process.

Parameters:

• recursive – if True, try to kill the complete process tree

• timeout – wait timeout (in seconds) after sending the kill signal (when recursive=True)

interrupt() → None

Send SIGINT to the process, kill on Windows.

is_running() → bool

Check whether the process is running.

children() → list[Any]

Return list of child processes (using psutil).

class os.process.File(name: Any, mode: str = 'r')

Can be a PIPE, a file object.

get_command() → str | bytes | None

Return the command to run to create the pipe.

close() → None

Close the file if needed.

exception os.process.WaitError

Bases: Exception

Common base class for all non-exit exceptions.

os.process.wait_for_processes(process_list: list[Run], timeout: float) → int | None

Wait for several processes spawned with Run.

Parameters:

• process_list – a list of Run objects

• timeout – a timeout in seconds. If 0 block until a process ends.

Returns:

None in case of timeout or the index in process Run corresponding to the first process that end

os.process.is_running(pid: int) → bool

Check whether a process with the given pid is running.

Parameters:

pid – an integer (e.g the value of Run().pid)

os.process.kill_process_tree(pid: int | Any, timeout: int = 3) → bool

Kill a hierarchy of processes.

Parameters:

• pid – pid of the toplevel process

• timeout – wait timeout after sending the kill signal

Returns:

True if all processes either don’t exist or have been killed, False if there are some processes still alive.