Last active
April 13, 2019 16:12
-
-
Save soxofaan/0bc37fcb122396321523d0ce5eb0fcd5 to your computer and use it in GitHub Desktop.
Python (3.6+) context manager that catches signals (e.g. SIGINT, SIGTERM) and exposes that. Allows writing infinite loops that can be exited in a clean way.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
import os | |
import signal | |
import time | |
log = logging.getLogger(__name__) | |
class SignalCatcher: | |
""" | |
Context manager that catches signals (e.g. SIGINT, SIGTERM) and exposes whether one was caught. | |
Allows to write an endless loop that can be exited cleanly | |
by sending SIGINT (e.g. keyboard interrupt) or SIGTERM (kill) | |
""" | |
def __init__(self, catch_message: str = 'Caught signal {signal!r}', signals: list = None): | |
self._caught = None | |
self._catch_message = catch_message | |
self._signals = signals or [signal.SIGINT, signal.SIGTERM] | |
self._origs = {} | |
def _handler(self, sig, frame): | |
sig = signal.Signals(sig) | |
log.warning(self._catch_message.format(signal=sig, name=sig.name, code=sig.value)) | |
self._caught = sig | |
def caught(self): | |
return self._caught | |
def __enter__(self): | |
# Set our custom handler (and keep originals to restore later) | |
for s in self._signals: | |
self._origs[s] = signal.getsignal(s) | |
signal.signal(s, self._handler) | |
return self | |
def __exit__(self, exc_type, exc_val, exc_tb): | |
# Restore original handlers | |
for s, orig in self._origs.items(): | |
signal.signal(s, orig) | |
def signal_catcher_example(): | |
""" | |
Usage example of `SignalCatcher`: run this in one terminal and interrupt it with keyboard interrupt | |
or with `kill` from another terminal | |
""" | |
logging.basicConfig(level=logging.INFO) | |
with SignalCatcher(catch_message='Caught {signal!r}, will stop looping now') as signal_catcher: | |
print("Hi, I'm process", os.getpid()) | |
print("Starting infinite loop. Interrupt me, but give me second to finish properly.") | |
while not signal_catcher.caught(): | |
print("start iteration") | |
time.sleep(3) | |
print("end iteration") | |
print("Terminated by", signal_catcher.caught()) | |
if __name__ == '__main__': | |
signal_catcher_example() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment