Skip to content

Instantly share code, notes, and snippets.

@joshbode
Forked from cshtdd/interruptable_region.py
Last active August 29, 2015 13:57
Show Gist options
  • Save joshbode/9781107 to your computer and use it in GitHub Desktop.
Save joshbode/9781107 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python
import signal
class InterruptableRegion(object):
def __init__(self, sig=signal.SIGINT):
self.sigs = list(sig) if isinstance(sig, (list, tuple)) else [sig]
self.orig_handlers = {}
self.interrupted = False
self.released = False
def _validate_region_start(self):
if self.interrupted or self.released or self.orig_handlers:
raise RuntimeError("An interruptable region can only be used once")
def _capture_orig_handlers(self):
self.orig_handlers = {
sig: signal.getsignal(sig)
for sig in self.sigs
}
def _restore_orig_handlers(self):
for sig, handler in self.orig_handlers.items():
signal.signal(sig, handler)
self.orig_handlers = {}
def _release(self):
if not self.released:
self._restore_orig_handlers()
self.released = True
def __enter__(self):
self._validate_region_start()
self._capture_orig_handlers()
def handler(signum, frame):
self._release()
self.interrupted = True
for sig in self.sigs:
signal.signal(sig, handler)
return self
def __exit__(self, type, value, tb):
self._release()
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