Django manage.py Command KeyboardInterrupt cleanup code execution

do you have an idea how to use signals in custom management commands in Django?

My Command's handle method processes continious stuff which needs to be cleaned up, when Command is interrupted.

Without any specific additional signal handling, my Command class could look like this:

class Command(BaseCommand):

    def handle(self, *args, **kwargs):
        while True:
            print("do continuous stuff")
            time.sleep(1)

    def cleanup(self, *args, **kwargs):
        print("do cleanup after continuous stuff is interrupted")

Does not work out, because signal is already caught by django:

    def handle(self, *args, **kwargs):
        try:
            while True:
                print("do continuous stuff")
                time.sleep(1)

        except KeyboardInterrupt as e:
            self.cleanup()
            raise e

Does not work out, because signal is not passed here:

    def handle(self, *args, **kwargs):
        self.is_interrupted = False
        signal.signal(signal.SIGINT, self.cleanup)
        while not self.is_interrupted:
            print("do continuous stuff")
            time.sleep(1)

    def cleanup(self, *args, **kwargs):
        self.is_interrupted = True
        print("do cleanup after continuous stuff is interrupted")
        raise KeyboardInter

Do you have an idea?

According to the hint of Abdul I checked if I could catch any signal of signal.valid_signals() on hitting Ctl + C and found out, it is not signal.SIGINT but signal.SIGTERM at least on my system, I can catch (Docker container based on python:latest image). With this information, this works for me:

import signal
import time
from django.core.management.base import BaseCommand

class Command(BaseCommand):

    def handle(self, *args, **kwargs):
        self.is_interrupted = False
        signal.signal(signal.SIGTERM, self.cleanup)

        while not self.is_interrupted:
            print("do continuous stuff")
            time.sleep(1)


    def cleanup(self, signal, frame):
        self.is_interrupted = True
        print("do cleanup after signal %s occoured" % signal)

Thank you very much!

Вернуться на верх