interruptable_region.py

#!/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()