Django manage.py Команда на клавиатуре прерывает выполнение кода очистки

у вас есть идея, как использовать сигналы в пользовательских командах управления в Django?

Метод handle моей команды обрабатывает непрерывный материал, который необходимо очистить, когда команда прерывается.

Без какой-либо специальной дополнительной обработки сигналов мой командный класс мог бы выглядеть следующим образом:

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")

Не получается, потому что сигнал уже пойман django:

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

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

Не получается, потому что здесь не передается сигнал:

    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

У тебя есть идея?

По подсказке Абдула я проверил, могу ли я уловить какой-либо сигнал signal.valid_signals() при нажатии Ctl + C и обнаружил, что это не signal.SIGINT, а signal.SIGTERM по крайней мере, в моей системе я могу перехватить (контейнер Docker на основе python: последнее изображение). С учетом этой информации, у меня это работает:

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)

Большое вам спасибо!

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