Module sppas.config

Class sppasExecProcess

Description

A convenient class to execute a subprocess.

A sppasExecProcess is a wrapper of 'subprocess.Popen' command. The sppasExecProcess class is a convenient wrapper for executing subprocesses in Python. It provides methods for running a command, getting the stdout and stderr of the command, stopping a running command, and checking the status of the command.

The main functionalities of the sppasExecProcess class are:

Running a command using 'subprocess.Popen' or 'subprocess.run'
Getting the stdout and stderr of the executed command
Stopping a running command
Checking the status of the executed command

Example

 >>> # Launch a command:
 >>> p = sppasExecProcess()
 >>> p.run_popen("ls -l")
 >>> # Get the stdout of the command:
 >>> p.out()
 >>> # Get the stderr of the command:
 >>> p.error()
 >>> # Stop a command:
 >>> p.stop()
 >>> # Get the state of the command:
 >>> p.is_running()

Constructor

Create a new instance.

View Source

def __init__(self):
    """Create a new instance."""
    self.__process = None

Public functions

test_command

Return True if command exists.

Test only the main command: the first string without its arguments.

Parameters

command: (str) a system command possibly with args.

Returns

(bool) True if the given command can be executed.

View Source

@staticmethod
def test_command(command):
    """Return True if command exists.

        Test only the main command: the first string without its arguments.

        :param command: (str) a system command possibly with args.
        :return: (bool) True if the given command can be executed.

        """
    command_args = shlex.split(command)
    test_command = command_args[0]
    logging.debug('Test of the command: {:s}'.format(test_command))
    fd_null = open(os.path.devnull, 'w')
    try:
        p = subprocess.Popen([test_command], shell=False, stdout=fd_null, stderr=fd_null)
    except OSError:
        fd_null.close()
        return False
    pid = p.pid
    p.terminate()
    try:
        os.kill(pid, signal.SIGINT)
        p.kill()
    except OSError:
        pass
    fd_null.close()
    return True

run_popen

Execute the given command with 'subprocess.Popen'.

Parameters

command: (str) The command to be executed

View Source

def run_popen(self, command):
    """Execute the given command with 'subprocess.Popen'.

        :param command: (str) The command to be executed

        """
    logging.info('Process command: {}'.format(command))
    command = command.strip()
    command_args = shlex.split(command)
    pipe = subprocess.PIPE
    self.__process = subprocess.Popen(command_args, stdout=pipe, stderr=pipe)
    self.__process.wait()

run

Execute command with 'subprocess.run'.

Requires Python 3.5+.

Parameters

command: (str) The command to execute
timeout: (int) Expiration delay in milliseconds

Raises

TimeoutExpired() after timeout seconds

View Source

def run(self, command, timeout=300):
    """Execute command with 'subprocess.run'.

        Requires Python 3.5+.

        :param command: (str) The command to execute
        :param timeout: (int) Expiration delay in milliseconds
        :raises: TimeoutExpired() after timeout seconds

        """
    logging.info('Process command: {}'.format(command))
    command = command.strip()
    command_args = shlex.split(command)
    try:
        self.__process = subprocess.run(command_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, cwd=None, timeout=timeout, check=False, encoding=None, errors=None, env=None, universal_newlines=True)
    except AttributeError:
        self.__process = subprocess.call(command_args, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False)

out

Return the standard output of the processed command.

Returns

(str) output message

View Source

def out(self):
    """Return the standard output of the processed command.

        :return: (str) output message

        """
    if self.__process is None:
        return ''
    out = self.__process.stdout
    out = str(out)
    return out

error

Return the error output of the processed command.

Returns

(str) error message

View Source

def error(self):
    """Return the error output of the processed command.

        :return: (str) error message

        """
    if self.__process is None:
        return ''
    error = self.__process.stderr
    error = str(error)
    return error

stop

Terminate the current command if it is running.

View Source

def stop(self):
    """Terminate the current command if it is running."""
    if self.is_running() is True:
        self.__process.terminate()

status

Return the status of the command if the process is completed.

Returns

(int) -2 means no process or process returned code

View Source

def status(self):
    """Return the status of the command if the process is completed.

        :return: (int) -2 means no process or process returned code

        """
    if self.__process is None:
        return -2
    return self.__process.returncode

is_running

Return True if the process is still running.

Returns

(bool)

View Source

def is_running(self):
    """Return True if the process is still running.

        :return: (bool)

        """
    if self.__process is None:
        return False
    return self.__process.poll() is None