Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How to make a python context manager catch a SIGINT or SIGTERM signal

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 ?

like image 301
gohu Avatar asked Jun 29 '20 16:06

gohu


People also ask

How do you capture a SIGINT in Python?

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.

Can SIGTERM be caught?

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.

How does Python handle SIGTERM?

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.

How do you send and receive signals in Python?

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.


1 Answers

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.

like image 143
nehtor.t Avatar answered Oct 19 '22 10:10

nehtor.t