dlpt.proc module

Functions for spawning, killing and getting process info.

exception dlpt.proc.SubprocError(cmd: str, returncode: int, stdout: str | None = None, stderr: str | None = None)

Bases: SubprocessError

Same as subprocess.CalledProcessError, but does not swallow stderr.

Note

stdout is silently swallowed (as with subprocess.SubprocessError), since processes stdout can be long and exception can be unreadable.

__init__(cmd: str, returncode: int, stdout: str | None = None, stderr: str | None = None)

exception dlpt.proc.SubprocTimeoutError(cmd: str, timeout_sec: float, stdout: str | None = None, stderr: str | None = None)

Bases: SubprocessError

Same as subprocess.TimeoutExpired, but does not swallow stderr.

Note

stdout is silently swallowed (as with subprocess.SubprocessError), since processes stdout can be long and exception can be unreadable.

__init__(cmd: str, timeout_sec: float, stdout: str | None = None, stderr: str | None = None)

dlpt.proc.get_name(pid: str | int) → str

Get process name.

Note

No PID existence check is performed.

Parameters:: pid – PID number.
Returns:: Process name as string.

dlpt.proc.get_executable(pid: str | int) → str

Get process executable path.

Note

No PID existence check is performed.

Parameters:: pid – PID number.
Returns:: Process executable as string.

dlpt.proc.get_cmd_args(pid: str | int) → List[str]

Return a list of process command line arguments as it was intially spawned.

Note

No PID existence check is performed.

Parameters:: pid – PID number.
Returns:: A list of process command line spawn arguments.

dlpt.proc.is_alive(pid: str | int | None) → bool

Return True if PID exists and process is running, False otherwise. Raise exception if given PID is None.

Parameters:: pid – PID number, string or integer.
Returns:: True if given PID exists and is running, False otherwise.

dlpt.proc.get_childs(pid: str | int) → List[int]

Return a list of child processes PIDs.

Note

No PID existence check is performed.

Parameters:: pid – PID number of a parent process, string or integer.
Returns:: A list of process child processes PIDs.

dlpt.proc.kill(pid: str | int, raise_exception: bool = True, timeout_sec: int | None = 3) → bool

Kill process with a given PID.

Parameters:

pid – PID number.
raise_exception – if True, exception is raised if process wasn’t successfully killed. Otherwise return False.
timeout_sec – wait for specified number of seconds for a process to be killed. If None, return immediately.

Returns:

True on successfully terminated process, False otherwise (or exception, based on raise_exception input argument).

dlpt.proc.kill_childs(pid: str | int, raise_exception: bool = True) → List[int]

Kill all child processes of a process with a given PID.

Parameters:

pid – PID number, string or integer. Raise Exception if None.
raise_exception – if True, exception is raised if any of child processes can’t be killed.

Returns:

A list of killed processes.

dlpt.proc.kill_tree(pid: str | int, raise_exception: bool = True) → List[int]

Kill parent process and all child processes.

Note

First, child processes are killed, than parent process.

Parameters:

pid – PID number, string or integer. Raise Exception if None.
raise_exception – if True, exception is raised if any of processes can’t be killed. Otherwise, False is returned.

Returns:

A list of killed processes.

dlpt.proc.kill_tree_multiple(pids: Sequence[str | int], raise_exception: bool = True) → List[int]

Iterate over given pids and perform kill_tree().

Parameters:

pids – a list of PIDs - string or integer. Raise Exception if None.
raise_exception – if True, exception is raised if any of processes can’t be killed.

Returns:

A list of killed processes.

dlpt.proc.get_alive(name_filter: str) → List[int]

Return a list of currently alive process PIDs.

Parameters:: name_filter – filter return list by finding name_filter in a process name (lower case string is compared).

Example

>>> dlpt.proc.get_alive("python.exe")
[26316, 33672, 73992] # pids of local running python.exe processes

Returns:: A list of currently alive process PIDs.

dlpt.proc.spawn_non_blocking_subproc(args: Sequence[str | int]) → int

Spawn non-blockin subprocess with given command line arguments and return PID.

Note

If spawned subprocess throw: “OSError: [WinError 740] The requested operation requires elevation” user does not have permission for executing them. Try to re-run script with admin permissions.

Parameters:: args – list of subprocess arguments.

Example

>>> args = ['python.exe', 'proc.py']
>>> dlpt.proc.spawn_non_blocking_subproc(args)
1234

Returns:: Spawned process PID.

dlpt.proc.spawn_subproc(args: Sequence[str | int], check_return_code: bool = True, stdout: int | None = -1, stderr: int | None = -1, stdin: int | None = -1, encoding: str = 'utf-8', timeout_sec: float | None = None, **run_args) → CompletedProcess

Spawn subprocess and return subprocess.CompletedProcess or raise exception. By default, raise exception on timeout (if given) or if return code is not zero. With **run_args, allow setting all subprocess.run() arguments.

Note

If spawned subprocess throw: “OSError: [WinError 740] The requested operation requires elevation” user does not have permission for executing them. Try to re-run script with admin permissions.

Parameters:

args – command line arguments with which process will be spawned. Can be shell commands, like ping. Note: all commandline arguments (specifically paths) must be properly encoded. For example, path containing tilde will throw error.
check_return_code – if True, return code is checked by run() function. In case it is not zero, SubprocError is raised. If False, subprocess.CompletedProcess is returned.
stdout – STDOUT routing specifier.
stderr – STDERR routing specifier.
stdin – STDIN routing specifier. Note: By default, ‘stdin’ is set to subprocess.PIPE, which should raise exception if spawned subprocess require user input.
encoding – STDOUT/ERR string encoding
timeout_sec – timeout in seconds. If None, no timeout is implemented. Else, if timeout is reached, process is killed and TimeoutExpired exception re-raised.
run_args – optional key-worded subprocess.run() arguments. Note: for the common basic subprocess.run() args, see spawn_subproc().

Example

>>> args = ['python.exe', 'proc.py']
>>> env_vars = {**os.environ, 'TEST_KEY': 'testenvvar'} # optional kwarg
>>> proc = dlpt.proc.spawn_subproc(args, timeout_sec: 3, env=env_vars)
>>> proc.pid
1234
>>> proc.returncode
0

Returns:: CompleteProcess object once process execution has finished or was terminated.

dlpt.proc.spawn_shell_subproc(args: Sequence[str | int], check_return_code: bool = True, encoding: str = 'utf-8', timeout_sec: float | None = None, **run_args) → CompletedProcess

Similar to spawn_subproc() but for shell commands. STDOUT/ERR is hidden from user and only set in returned proc.stdout/err.

Note

If spawned subprocess throw: “OSError: [WinError 740] The requested operation requires elevation” user does not have permission for executing them. Try to re-run script with admin permissions.

Parameters:

args – command line arguments with which process will be spawned. Can be shell commands, like ping. Note: all commandline arguments (specifically paths) must be properly encoded. For example, path containing tilde will throw error.
check_return_code – if True, return code is checked by run() function. In case it is not zero, SubprocError is raised. If False, subprocess.CompletedProcess is returned.
encoding – STDOUT/ERR string encoding
timeout_sec – timeout in seconds. If None, no timeout is implemented. Else, if timeout is reached, process is killed and TimeoutExpired exception re-raised.
run_args – optional key-worded subprocess.run() arguments, that are added to run() call. Note: for the common, basic subprocess.run() args, see spawn_subproc()

Example

>>> args = ['dir']
>>> dlpt.proc.spawn_shell_subproc(args)
proc.py, pth.py, # ...

Returns:: CompleteProcess object once process execution has finished or was terminated.