Last active — forked from nonZero/graceful_int_handler.py

Embed URL

HTTPS clone URL

SSH clone URL

You can clone with HTTPS or SSH.

Download Gist
View interruptable_region.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76
import signal
 
class InterruptableRegion(object):
def __init__(self, sig=signal.SIGINT):
self.sig = sig
self.interrupted = False
self.released = False
self.original_handler = None
 
def __enter__(self):
self._validate_region_start()
self._store_signal_default_handler()
 
def _signal_invoked_new_handler(signum, frame):
self._release()
self.interrupted = True
 
signal.signal(self.sig, _signal_invoked_new_handler)
 
return self
 
def __exit__(self, type, value, tb):
self._release()
 
def _validate_region_start(self):
if self.interrupted or self.released or self.original_handler:
raise RuntimeError("An interruptable region can only be used once")
 
def _release(self):
if not self.released:
self._reset_signal_default_handler()
self.released = True
 
def _store_signal_default_handler(self):
self.original_handler = signal.getsignal(self.sig)
 
def _reset_signal_default_handler(self):
signal.signal(self.sig, self.original_handler)
 
 
if __name__ == '__main__':
 
import unittest
import time
class InterruptableRegionTestCase(unittest.TestCase):
def test_simple(self):
with InterruptableRegion() as h:
while True:
print "..."
time.sleep(1)
if h.interrupted:
print "interrupted!"
time.sleep(5)
break
def test_nested(self):
with InterruptableRegion() as h1:
while True:
print "(1)..."
time.sleep(1)
with InterruptableRegion() as h2:
while True:
print "\t(2)..."
time.sleep(1)
if h2.interrupted:
print "\t(2) interrupted!"
time.sleep(2)
break
if h1.interrupted:
print "(1) interrupted!"
time.sleep(2)
break
unittest.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.