signals.py

importsignal
importweakref

fromfunctoolsimportwraps

__unittest=True


class_InterruptHandler(object):
def__init__(self, default_handler):
self.called=False
self.original_handler=default_handler
ifisinstance(default_handler, int):
ifdefault_handler==signal.SIG_DFL:
# Pretend it's signal.default_int_handler instead.
default_handler=signal.default_int_handler
elifdefault_handler==signal.SIG_IGN:
# Not quite the same thing as SIG_IGN, but the closest we
# can make it: do nothing.
defdefault_handler(unused_signum, unused_frame):
pass
else:
raiseTypeError("expected SIGINT signal handler to be "
"signal.SIG_IGN, signal.SIG_DFL, or a "
"callable object")
self.default_handler=default_handler

def__call__(self, signum, frame):
installed_handler=signal.getsignal(signal.SIGINT)
ifinstalled_handlerisnotself:
# if we aren't the installed handler, then delegate immediately
# to the default handler
self.default_handler(signum, frame)

ifself.called:
self.default_handler(signum, frame)
self.called=True
forresultin_results.keys():
result.stop()

_results=weakref.WeakKeyDictionary()
defregisterResult(result):
_results[result] =1

defremoveResult(result):
returnbool(_results.pop(result, None))

_interrupt_handler=None
definstallHandler():
global_interrupt_handler
if_interrupt_handlerisNone:
default_handler=signal.getsignal(signal.SIGINT)
_interrupt_handler=_InterruptHandler(default_handler)
signal.signal(signal.SIGINT, _interrupt_handler)


defremoveHandler(method=None):
ifmethodisnotNone:
@wraps(method)
definner(*args, **kwargs):
initial=signal.getsignal(signal.SIGINT)
removeHandler()
try:
returnmethod(*args, **kwargs)
finally:
signal.signal(signal.SIGINT, initial)
returninner

global_interrupt_handler
if_interrupt_handlerisnotNone:
signal.signal(signal.SIGINT, _interrupt_handler.original_handler)