Skip to content

Instantly share code, notes, and snippets.

@silviud
Last active June 15, 2016 13:47
Show Gist options
  • Save silviud/7641569 to your computer and use it in GitHub Desktop.
Save silviud/7641569 to your computer and use it in GitHub Desktop.
Python sub-process with timeout
from subprocess import PIPE, Popen
from signal import alarm, signal, SIGALRM, SIGKILL
import logging
from traceback import format_exception
import sys
class RuntimeException(Exception):
pass
class Bash:
def __init__(self, args, timeout=600):
self.args = args
logging.debug("execute:%s"%args)
self.timeout = int(timeout)
self.process = None
self.success = False
self.run()
def run(self):
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
try:
self.process = Popen(self.args, shell=True, stdout=PIPE, stderr=PIPE)
if self.timeout != -1:
signal(SIGALRM, alarm_handler)
alarm(self.timeout)
try:
self.stdout, self.stderr = self.process.communicate()
if self.timeout != -1:
alarm(0)
except Alarm:
os.kill(self.process.pid, SIGKILL)
raise RuntimeException("Timeout during command execution")
self.success = self.process.returncode == 0
except:
raise RuntimeException(format_exception(*sys.exc_info()))
if not self.success:
logging.debug("Failed to execute:" + self.get_err_msg())
def is_success(self):
return self.success
def get_stdout(self):
return self.stdout.strip("\n")
def get_lines(self):
return self.stdout.split("\n")
def get_stderr(self):
return self.stderr.strip("\n")
def get_err_msg(self):
if self.is_success():
return ""
if self.get_stderr() is None or self.get_stderr() == "":
return self.get_stdout()
else:
return self.get_stderr()
if __name__ == '__main__':
b = Bash('ls -al /etc/')
b.get_lines()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment