如何在Django自定义管理命令中用信号处理中断后的清理？

问题背景 #

需要在Django自定义管理命令的持续任务被中断（比如按Ctrl+C）时自动执行清理操作，之前尝试的两种方案均失效：

• 直接捕获KeyboardInterrupt：Django管理命令框架会在handle方法外层捕获该异常，导致内部的try...except无法触发。
• 绑定SIGINT信号到cleanup方法：代码存在拼写错误，且未保留原始信号处理逻辑，导致信号无法正确传递，命令无法正常退出。

可行解决方案 #

方案：自定义信号处理+保留原始信号逻辑 #

通过保存Django默认的SIGINT处理函数，在自定义信号处理中先执行清理，再调用原始处理逻辑，既完成清理又保证命令正常退出：

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

class Command(BaseCommand):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.interrupted = False
        # 保存Django默认的SIGINT处理函数
        self.original_sigint = signal.getsignal(signal.SIGINT)

    def handle(self, *args, **kwargs):
        # 绑定自定义的SIGINT处理函数
        signal.signal(signal.SIGINT, self.handle_sigint)
        try:
            while not self.interrupted:
                print("do continuous stuff")
                time.sleep(1)
        finally:
            # 恢复原始信号处理，避免影响其他逻辑
            signal.signal(signal.SIGINT, self.original_sigint)

    def handle_sigint(self, signum, frame):
        # 标记中断状态，终止主循环
        self.interrupted = True
        # 执行清理操作
        self.cleanup()
        # 调用原始SIGINT处理，确保命令正常退出
        self.original_sigint(signum, frame)

    def cleanup(self):
        print("do cleanup after continuous stuff is interrupted")

方案说明 #

• 初始化时保存原始SIGINT处理函数，避免完全覆盖Django的默认退出逻辑
• 自定义信号处理函数中先标记中断状态、执行清理，再调用原始处理函数，保证命令按Django流程正常退出
• finally块中恢复原始信号处理，防止对后续代码产生影响

补充场景方案 #

如果需要在任何退出场景（包括正常结束）都执行清理，可结合atexit模块注册清理函数：

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

class Command(BaseCommand):
    def handle(self, *args, **kwargs):
        atexit.register(self.cleanup)
        while True:
            print("do continuous stuff")
            time.sleep(1)

    def cleanup(self):
        print("do cleanup")

此方法会在命令正常结束时也执行清理，需根据实际需求选择。

内容的提问来源于stack exchange，提问作者ahandi