Мне кажется, я нашел подходящее решение: я создал собственное небольшое консольное приложение и подключился к его очереди сообщений, чтобы перехватить событие завершения работы. Я еще не тестировал его, и я также не знаю, является ли это хорошим решением, но, возможно, оно кому-нибудь пригодится.
Сначала приведу код для моей простой консоли, основанной на tkinter. Он показывает stdout в черном и stderr в красном:
# a simple console based on tkinter to display stdout and stderr
class SimpleConsole(object):
def __init__(self, name):
self.root = Tk()
self.root.title(name)
self.init_ui()
def init_ui(self):
self.text_box = Text(self.root, wrap='word', height = 11, width=50)
self.text_box.grid(column=0, row=0, columnspan = 2, sticky='NSWE', padx=5, pady=5)
self.text_box.tag_config('std', foreground="black")
self.text_box.tag_config('err', foreground="red")
self.text_box.pack(side=LEFT, fill=BOTH, expand = YES)
self.text_box.yview()
self.yscrollbar = Scrollbar(self.root, orient=VERTICAL, command=self.text_box.yview)
self.yscrollbar.pack(side=RIGHT, fill=Y)
self.text_box["yscrollcommand"] = self.yscrollbar.set
sys.stdout = SimpleConsole.StdRedirector(self.text_box, "std")
sys.stderr = SimpleConsole.StdRedirector(self.text_box, "err")
self.update()
class StdRedirector(object):
def __init__(self, text_widget, tag):
self.text_space = text_widget
self.tag = tag
def write(self, string):
self.text_space.insert('end', string, self.tag)
self.text_space.see('end')
def flush(self):
pass
def update(self):
self.root.update()
def get_window_handle(self):
return int(self.root.wm_frame(), 16)
Затем я создал класс, который подключается к очереди сообщений моей консоли и управляет выключением:
#class to handle a graceful shutdown by hooking into windows message queue
class GracefulShutdown:
def __init__(self, handle):
self.shutdown_requested = False
self._shutdown_functions = []
self.handle = handle
try:
if os.name == 'nt':
# Make a dictionary of message names to be used for printing below
self.msgdict = {}
for name in dir(win32con):
if name.startswith("WM_"):
value = getattr(win32con, name)
self.msgdict[value] = name
# Set the WndProc to our function
self.oldWndProc = win32gui.SetWindowLong(self.handle, win32con.GWL_WNDPROC, self.my_wnd_proc)
if self.oldWndProc == 0:
raise NameError("wndProc override failed!")
self.message_map = {win32con.WM_QUERYENDSESSION: self.hdl_query_end_session,
win32con.WM_ENDSESSION: self.hdl_end_session,
win32con.WM_QUIT: self.hdl_quit,
win32con.WM_DESTROY: self.hdl_destroy,
win32con.WM_CLOSE: self.hdl_close}
# pass a shutdown message to windows
retval = windll.user32.ShutdownBlockReasonCreate(self.handle,c_wchar_p("I'm still saving data!"))
if retval == 0:
raise NameError("shutdownBlockReasonCreate failed!")
except Exception as e:
logging.exception("something went wrong during win32 shutdown detection setup")
#catches all close signals and passes it to our own functions; all other signals are passed to the original function
def my_wnd_proc(self, hwnd, msg, w_param, l_param):
# Display what we've got.
logging.debug(self.msgdict.get(msg), msg, w_param, l_param)
# Restore the old WndProc. Notice the use of wxin32api
# instead of win32gui here. This is to avoid an error due to
# not passing a callable object.
if msg == win32con.WM_DESTROY:
win32api.SetWindowLong(self.handle,
win32con.GWL_WNDPROC,
self.oldWndProc)
#simplify function for calling
def call_window_proc_old():
return win32gui.CallWindowProc(self.oldWndProc, hwnd, msg, w_param, l_param)
#either call our handle functions or call the original wndProc
return self.message_map.get(msg, call_window_proc_old)()
def hdl_query_end_session(self):
logging.info("WM_QUERYENDSESSION received")
self.shutdown_requested = True
#we have to return 0 here to prevent the windows shutdown until our application is closed
return 0
def hdl_end_session(self):
logging.info("WM_ENDSESSION received")
self.exit_gracefully()
return 0
def hdl_quit(self):
logging.info("WM_QUIT received")
self.shutdown_requested = True
return 0
def hdl_destroy(self):
logging.info("WM_DESTROY received")
return 0
def hdl_close(self):
logging.info("WM_CLOSE received")
self.shutdown_requested = True
return 0
def exit_gracefully(self):
logging.info("shutdown request received")
self.shutdown_requested = True
for func in self._shutdown_functions:
try:
func()
except:
logging.exception("Exception during shutdown function:")
logging.info("shutdown request done, bye!")
exit(0)
def add_cleanup_function(self, function):
self._shutdown_functions.append(function)
И вотнекоторый «основной» код для запуска обоих классов и его проверки:
if __name__ == "__main__":
import time
from logging.handlers import RotatingFileHandler
#setup own console window
console = SimpleConsole("Test Shutdown")
#setup 3 loggers:
#log debug and info to stdout
#log warning and above to stderr
#log info and above to a file
logging.getLogger().setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging_path = 'graceful_shutdown_test.log'
rot_file_handler = RotatingFileHandler(logging_path, maxBytes=50 * 1024 * 1024, backupCount=5)
rot_file_handler.setFormatter(formatter)
rot_file_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(rot_file_handler)
log_to_stdout = logging.StreamHandler(sys.stdout)
log_to_stdout.setLevel(logging.INFO)
log_to_stdout.addFilter(lambda record: record.levelno <= logging.INFO)
log_to_stdout.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stdout)
log_to_stderr = logging.StreamHandler()
log_to_stderr.setLevel(logging.WARNING)
log_to_stderr.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stderr)
logging.info("start shutdown test")
#init graceful shutdown with tkinter window handle
shutdown = GracefulShutdown(console.get_window_handle())
counter = 0
counterError = 0
#test cleanup function which runs if shutdown is requested
def graceful_shutdown():
logging.info("start shutdown")
time.sleep(15)
logging.info("stop shutdown")
shutdown.add_cleanup_function(graceful_shutdown)
#main test loop
while not shutdown.shutdown_requested:
console.update()
counter += 1
if counter > 50:
logging.info("still alive")
counter = 0
counterError += 1
if counterError > 150:
logging.error("error for test")
try:
raise NameError("i'm a exception")
except:
logging.exception("exception found!")
counterError = 0
time.sleep(0.1)
shutdown.exit_gracefully()