SPPAS 4.20

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