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

Question

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 ?

Swink

Does this answer your question? How do I capture SIGINT in Python? – Theologue

Not exactly, I have updated the question in that regard – Swink

Doesn’t your context manager already work for SIGINT? – Landholder

You need to register whichever function you want to be called as a signal handler, for example self.close. – Theologue

Yes, and it might already work as you expect without doing that because it’s a context manager. Can you show a minimal reproducible example where the script does not work as you expect when you use the kill command? – Theologue

I have tried some stuff using signal on my side. What makes it more complex is that I use threading to stream and write the data. I will update my answer with what works so far and see if there is a better idea to try – Swink

Answer 1

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.

Witt

How could this be used with docs.python.org/3/library/…? – Retouch

I guess you could use the above code and extend ContextDecorator, which will allow you to use it as decorator with any function – Witt