5
\$\begingroup\$

The class below provides a set of utility methods to currently running process. It is a wrapper around Python PSUtil to check status of a process in localhost as well as remote host. Please help me improve this.

from   collections import namedtuple
import json
import psutil

class ProcessStatus(object):
    """
    This class provides a set of utility methods to access a process.
    """

    # Currently this class exposes only these attributes from psutil.Process.
    _as_dict_attrnames = ['username', 'pid', 'ppid', 'status', 'terminal', 'create_time', 'cmdline', 'environ']

    # Namedtuple to represent a process.
    Process = namedtuple('Process', _as_dict_attrnames)

    @staticmethod
    def _get_process_retrieval_command_line(pid):
        """
        Returns python one liner command to find process with given pid. This is used to find a process on remote host
        by launching the returned command on the remote host.
        """
        return "python -c 'from myapp.util.process_status import ProcessStatus; import json;" \
               " print(json.dumps(ProcessStatus._find_local_process(pid={pid})))'".format(pid=pid)

    @staticmethod
    def _find_local_process(pid):
        """
        Returns a process with given pid running on localhost.
        """
        process = None
        try:
            process = psutil.Process(pid=pid).as_dict(attrs=ProcessStatus._as_dict_attrnames)
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            return process

        return process

    @staticmethod
    def find_process(pid, host='localhost'):
        """
        Returns a process with given pid running on given host.

        :param pid: pid of the process.
        :param host: host on which process is running.

        :rtype: C{ProcessStatus.Process}
        """

        # ToDo: Add validation for host provided to be a valid host string.
        if not isinstance(pid, int):
            raise TypeError(TYPE_MISMATCH_ERROR_MESSAGE.format(actual=type(pid), expected=int))

        if not isinstance(host, str):
            raise TypeError(TYPE_MISMATCH_ERROR_MESSAGE.format(actual=type(host), expected=str))

        process = None
        if is_local_host(host):
            process = ProcessStatus._find_local_process(pid=pid)
        else:
            # Getting python online command to retrieve process detail on local host
            remote_process_retrieval_command = ProcessStatus._get_process_retrieval_command_line(pid=pid)

            # Capturing process details from remote host by launching python one liner.
            return_code, process_output = execute_remote_command(command=remote_process_retrieval_command, host=host)

            if not return_code:
                # deserializing output of process from json to python dict.
                process = json.loads(process_output)

        if process is not None:
            return ProcessStatus.Process(**process)

        return process

    @staticmethod
    def find_descendant_process(pid, name):
        """
        Returns a process with given name in the process tree of process with give pid on local host.

        :param pid: Pid of the process which is to be considered as root of the process tree.
        :param name: Name of the descendant process.
        :rtype: C{ProcessStatus.Process}
        """
        if not isinstance(pid, int):
            raise TypeError(TYPE_MISMATCH_ERROR_MESSAGE.format(actual=type(pid), expected=int))

        if not isinstance(name, str):
            raise TypeError(TYPE_MISMATCH_ERROR_MESSAGE.format(actual=type(name), expected=str))

        process = None

        current_process = None
        try:
            current_process = psutil.Process(pid=pid)
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            return process

        if not current_process:
            return process

        # iterate over all descendant of the given process
        for child_process in current_process.children(recursive=True):
            if name == child_process.name():
                process = child_process.as_dict(attrs=ProcessStatus._as_dict_attrnames)
                break

        if process is not None:
            return ProcessStatus.Process(**process)

        return process
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.