1#@!#!123s

��JFIF�� 1#@!#!123s
: / opt / saltstack / salt / extras-3.10 / cproc /
Filename : process.py
back
"""cproc module process execution objects"""
import signal
import logging
from multiprocessing import cpu_count
from typing import Callable, Union
import subprocess as s
import rads.vz
from .monitor import ProcMonitor


class ProcLimit:
    """Contains load limits for cproc.Proc if used as the lim= arg

    Note:
        The grace argument could be useful for subprocesses which require a TCP
        session, such as rsync, to avoid a timeout caused by the process
        sleeping for too long. ex: ProcLimit(grace=(5, 1))

    Args:
        value (int | float | None): maximum sever load. If set to
            None, it'll default to either the number of cores determined from
            multiprocessing.cpu_count(), or 1.0 if running on a VZ VPS
        grace (tuple[int, int] | None): If set, it should be a 2-item
            tuple of ints > 0. The first item is the number of seconds
            to sleep if the server is overloaded, and the second is how
            long to allow execution anyways once the sleep expires
        max_mem (int | None): If set, the process will be killed if it uses
            more than this amount of memory, in MiB
        mem_signal (Signal | int): kill signal to send the process if it goes
            over max_mem. Defaults to ``signal.SIGTERM``
        mem_log_func (Callable | None): log function to call if a process is
            killed for being over memory. Defaults to ``logging.error``
    """

    __module__ = 'cproc'

    def __init__(
        self,
        value: Union[int, float, None] = None,
        grace: Union[tuple[int, int], None] = None,
        max_mem: Union[int, None] = None,
        mem_signal: Union[int, signal.Signals] = signal.SIGTERM,
        mem_log_func: Union[Callable, None] = logging.error,
    ) -> None:
        # memory limits MB -> B
        self.max_mem = max_mem * 2 ** 20 if max_mem else None
        self.mem_signal = mem_signal
        self.mem_log_func = mem_log_func
        if value is None:
            value = 1.0 if rads.vz.is_vps() else cpu_count()
        self.value = value
        self.grace = grace


class Proc(s.Popen):
    """Custom subprocess.Popen subclass which allows automatic pausing and
    resuming based on server load and has no defaults for encoding, stdout,
    and stderr.

    Warning:
        Using None for stdout or stderr can be dangerous if your program runs as
        a daemon under systemd. Systemd can kill the subprocess with SIGPIPE.
        It's better to set the outputs to Proc.DEVNULL if you don't need them.

    Args:
        args: A string, or a sequence of program arguments
        lim (int, float, ProcLimit | None): max load limit. This can be a static
            limit if you set it to a float or int, or for more flexibility,
            use a ``ProcLimit`` instance
        shell (bool): If true, the command will be executed through the
            shell (unsafe)
        encoding (str | None): text encoding for subprocess output and input,
            such as "UTF-8". Set this to None to use no encoding (as bytes)
        stdout (int | None): specifies the executed program's standard output.
            Set to Proc.DEVNULL to not capture output. or Proc.PIPE to capture.
            Use None to print to your shell, but see Warning above
        stderr (int | None): specifies the executed program's error output.
            Set to Proc.DEVNULL to not capture output. or Proc.PIPE to capture.
            Use None to print to your shell, but see Warning above
    """

    __module__ = 'cproc'

    # static variables for convenience; access them from Proc instead of having
    # to import each from subprocess too
    PIPE = s.PIPE
    DEVNULL = s.DEVNULL
    STDOUT = s.STDOUT
    TimeoutExpired = s.TimeoutExpired
    CalledProcessError = s.CalledProcessError
    CompletedProcess = s.CompletedProcess

    def __init__(
        self,
        args,
        *,
        lim: Union[int, float, ProcLimit, None],
        shell: bool = False,
        encoding: Union[str, None],
        stdout: Union[int, None],
        stderr: Union[int, None],
        **kwargs,
    ):
        if lim is None:
            self.lim = None
        elif isinstance(lim, ProcLimit):
            self.lim = lim
        else:
            self.lim = ProcLimit(lim)
        super().__init__(
            args,
            shell=shell,
            encoding=encoding,
            stdout=stdout,
            stderr=stderr,
            **kwargs,
        )
        if self.lim is not None:
            self.mon = ProcMonitor(self)

    # pylint: disable=redefined-builtin
    # input is a builtin, but also what vanilla subprocess calls it
    def complete(
        self,
        input: Union[bytes, str, None] = None,
        timeout: Union[int, float, None] = None,
        check: bool = False,
    ):
        """Wait for a process to complete and return a CompletedProcess instance

        Args:
            input (bytes | str | None): text to send to stdin.
                stdin should have already been set to PIPE before
                sending anything to it
            timeout (int | float | None): raise Proc.TimeoutExpired if the
                process takes longer than this number of seconds to execute.
                Defaults to None
            check (bool): if True, raise Proc.CalledProcessError if the process
                exits with a non-zero exit code. The raised object will
                contain the .returncode attribute. Defaults to False

        Raises:
            Proc.CalledProcessError: program exited with a non-zero exit code
                and check=True was specified
            Proc.TimeoutExpired: program took too long to execute and timeout=
                was specified

        Returns:
            Proc.CompletedProcess: contains .args, .returncode, .stdout,
            and .stdin
        """
        try:
            stdout, stderr = self.communicate(input=input, timeout=timeout)
        except:  # Includes KeyboardInterrupt; communicate handled that
            self.kill()
            self.wait()
            raise
        # communicate() already called self.wait()
        if check and self.returncode:
            raise Proc.CalledProcessError(
                self.returncode, self.args, output=stdout, stderr=stderr
            )
        return Proc.CompletedProcess(self.args, self.returncode, stdout, stderr)

    @staticmethod
    def run(
        args,
        *,
        lim,
        shell: bool = False,
        encoding: Union[str, None],
        input: Union[bytes, str, None] = None,
        capture_output: bool = False,
        timeout: Union[int, float, None] = None,
        check: bool = False,
        **kwargs,
    ) -> s.CompletedProcess:
        """Run command with arguments and return a CompletedProcess instance

        Args:
            args: A string, or a sequence of program arguments
            lim: max load limit. This can be a float, an int, an object with a
                .value property (such as a multiprocessing.Value or
                cproc.ProcLimit object), or an object with .__int__() defined
                to allow casting to an int
            shell (bool): If true, the command will be executed through the
                shell (unsafe)
            encoding (str | None): text encoding for subprocess output and
                input, such as "UTF-8". Set this to None to use no
                encoding (as bytes)
            input (bytes | str | None): text to send to stdin. Do not use the
                stdin= kwarg if you use input=
            capture_output (bool, optional): Sets stdout and stderr to
                Proc.PIPE if True. Do not also manually set stdout and stderr if
                this is set to True
            timeout (int | float | None): raise Proc.TimeoutExpired if the
                process takes longer than this number of seconds to execute.
                Defaults to None
            check (bool): if True, raise Proc.CalledProcessError if the process
                exits with a non-zero exit code. The raised object will
                contain the .returncode attribute. Defaults to False

        Raises:
            FileNotFoundError: requested program to execute does not exist
            Proc.CalledProcessError: program exited with a non-zero exit code
                and check=True was specified
            Proc.TimeoutExpired: program took too long to execute and timeout=
                was specified

        Returns:
            Proc.CompletedProcess: contains .args, .returncode, .stdout,
            and .stdin
        """
        if input is not None:
            if 'stdin' in kwargs:
                raise TypeError('stdin and input are mutually exclusive')
            kwargs['stdin'] = Proc.PIPE
        if capture_output:
            if 'stdout' in kwargs or 'stderr' in kwargs:
                raise TypeError(
                    'stdout and stderr cannot be manually '
                    'specified if capture_output=True'
                )
            stdout_num = Proc.PIPE
            stderr_num = Proc.PIPE
        else:
            try:
                stdout_num = kwargs.pop('stdout')
                stderr_num = kwargs.pop('stderr')
            except KeyError as key_exc:
                raise TypeError(
                    'stdout and stderr kwargs are required '
                    'if capture_output=False'
                ) from key_exc
        return Proc(
            args,
            lim=lim,
            shell=shell,
            stdout=stdout_num,
            stderr=stderr_num,
            encoding=encoding,
            **kwargs,
        ).complete(input=input, timeout=timeout, check=check)