I stream data using context manager to close the connection when the program exits. I run my program as a daemon in the background.
How can I make the context manager handle the case when the daemon is interrupted by a SIGINT or SIGTERM or any interrupt signal sent by the kill command ?
I am running Python 3 on a Raspberry Pi and Ubuntu.
I have seen this: How do I capture SIGINT in Python? Which is helpful, but I am not sure how to use that with python's context manager ? ie. let's say I have an object that I have built as a context manager:
class Sensor:
def __init__(self, name: str):
self.name = name
def __enter__(self):
self._connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
I use that object inside a script that is run as daemon. Is there a pythonic way to specify to specify that the __exit__
function has to be called also on SIGINT
and SIGTERM
exceptions ?
You can use the functions in Python's built-in signal module to set up signal handlers in python. Specifically the signal. signal(signalnum, handler) function is used to register the handler function for signal signalnum . Show activity on this post.
The SIGKILL or SIGSTOP signals cannot be caught or ignored. You can catch a signal in Linux by using sigaction . Use only functions that are async-signal-safe in the signal handler.
Python provides the Signal library allowing developers to catch Unix signals and set handlers for asynchronous events. For example, the 'SIGTERM' (Terminate) signal is received when issuing a 'kill' command for a given Unix process.
8.1: Send Signal to a Thread and Wait for it using sigwait() It then waits for the list of signals (SIGTERM and SIGALRM) using sigwait() method. Once the signal is received, it retrieves the handler of the signal and executes it. Our main code starts by registering a handler with SIGTERM signal.
You could create a internal method (e.g. _handle_interrupt
) for your context manager that calls sys.exit()
and register it as a signal handler once __enter__
is called:
class Sensor:
def __init__(self, name: str):
self.name = name
def _handle_interrupt(self):
sys.exit() # will trigger a exception, causing __exit__ to be called
def __enter__(self):
signal.signal(signal.SIGINT, self._handle_interrupt)
signal.signal(signal.SIGTERM, self._handle_interrupt)
self._connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
You should also have a look at PEP 419, which could be helpful for your setup. As for your requirements with regards to threading it is hard to tell without more information.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With