A collection of safe process and system utilities

!spawning.py

import logging
import typing

import psutil


T = typing.TypeVar("T")


def safe_execute(function: typing.Callable[..., T], *args, **kwargs, timeout: int = 60) -> T:
    """safely executes a function, even if it's spawning child processes.
    
    The function's child processes are terminated or killed even on function error.
    """
    try:
        return function(*args, **kwargs)
    finally:
        children = psutil.Process().children(recursive=True)

        for process in children:
            process.terminate()

        _, children = psutil.wait_procs(children, timeout=timeout)

        for process in children:
            message = f"The process with PID {process.pid} is being killed"
            logging.warning(message)
            
            process.kill()

~requirements.txt

psutil>7