Skip to content

Instantly share code, notes, and snippets.

@soxofaan
Last active April 13, 2019 16:12
Show Gist options
  • Save soxofaan/0bc37fcb122396321523d0ce5eb0fcd5 to your computer and use it in GitHub Desktop.
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.
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