pjf_executor.py 文件源码

def spawn(self, cmd, stdin_content="", stdin=False, shell=False, timeout=2):
        """
        Spawn a new process using subprocess
        """
        try:
            if type(cmd) != list:
                raise PJFInvalidType(type(cmd), list)
            if type(stdin_content) != str:
                raise PJFInvalidType(type(stdin_content), str)
            if type(stdin) != bool:
                raise PJFInvalidType(type(stdin), bool)
            self._in = stdin_content
            try:
                self.process = subprocess.Popen(cmd, stdout=PIPE, stderr=PIPE, stdin=PIPE, shell=shell)
                self.finish_read(timeout, stdin_content, stdin)
                if self.process.poll() is not None:
                    self.close()
            except KeyboardInterrupt:
                return
        except OSError:
            raise PJFProcessExecutionError("Binary <%s> does not exist" % cmd[0])
        except Exception as e:
            raise PJFBaseException(e.message if hasattr(e, "message") else str(e))