[Python] Unhandled exceptions from main thread in a multithread PyQt5 application

Discussão em 'Python' iniciado por Stack, Janeiro 22, 2025.

  1. Stack

    Stack Membro Participativo

    I'm struggling to find out a solution in order to manage unhandled exceptions from a main thread in a PyQt5 application, in a multithreading scenario. As long as the application is running in a single thread, I can catch every unhandled exceptions overriding sys.excepthook with a custom function. When a second thread starts, this is no longer possible. The same exceptions is not catched and the program crashes (while the worker keeps running). How can I prevent this? And how I can manage unhandled exceptions (from the main thread) in a multithreading scenario?

    Here a minimal code example reproducing the error:

    from PyQt5.QtCore import QThread, pyqtSignal
    from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QTextEdit, QMessageBox
    import logging
    import sys
    import traceback


    class WorkerThread(QThread):
    error_signal = pyqtSignal(str) # Signal to send errors to the main thread

    def __init__(self):
    super().__init__()
    self.running = True # Control flag

    def run(self):
    """Worker thread logic"""
    try:
    while self.running:
    print("Worker is running")
    self.msleep(1000) # Simulate work
    except Exception as e:
    error_message = f"WorkerThread Exception: {str(e)}"
    self.error_signal.emit(error_message)

    def stop(self):
    """Signal the thread to stop"""
    self.running = False
    self.quit()
    self.wait()


    class MainWindow(QMainWindow):
    def __init__(self):
    super().__init__()
    self.setWindowTitle("Worker Thread Exception Handling Example")
    self.resize(600, 400)

    # Worker thread setup
    self.worker_thread = None
    self.create_worker()

    # GUI setup
    self.text_edit = QTextEdit(self)
    self.text_edit.setReadOnly(True)

    self.start_button = QPushButton("Start Worker Thread", self)
    self.start_button.clicked.connect(self.start_worker_thread)

    self.stop_button = QPushButton("Stop Worker Thread", self)
    self.stop_button.clicked.connect(self.stop_worker_thread)

    self.throw_exception_button = QPushButton("Throw Exception in Main Thread", self)
    self.throw_exception_button.clicked.connect(self.throw_main_thread_exception)

    layout = QVBoxLayout()
    layout.addWidget(self.text_edit)
    layout.addWidget(self.start_button)
    layout.addWidget(self.stop_button)
    layout.addWidget(self.throw_exception_button)

    central_widget = QWidget(self)
    central_widget.setLayout(layout)
    self.setCentralWidget(central_widget)

    def create_worker(self):
    """Create a new worker thread"""
    if self.worker_thread and self.worker_thread.isRunning():
    self.stop_worker_thread()

    self.worker_thread = WorkerThread()
    self.worker_thread.error_signal.connect(self.handle_worker_error)

    def start_worker_thread(self):
    """Start the worker thread"""
    if self.worker_thread.isRunning():
    QMessageBox.warning(self, "Warning", "Worker thread is already running!")
    return

    self.text_edit.append("Worker thread started.")
    self.worker_thread.start()

    def stop_worker_thread(self):
    """Stop the worker thread"""
    if self.worker_thread and self.worker_thread.isRunning():
    self.text_edit.append("Stopping worker thread...")
    self.worker_thread.stop()
    self.text_edit.append("Worker thread stopped.")

    def handle_worker_error(self, error_message):
    """Handle errors from the worker thread"""
    self.text_edit.append(f"Worker Error: {error_message}")
    QMessageBox.critical(self, "Worker Thread Error", error_message)

    def throw_main_thread_exception(self):
    """Simulate an exception in the main thread"""
    raise ValueError("Simulated exception in the main thread")


    def exception_hook(exc_type, exc_value, exc_traceback):
    error_message = "".join(traceback.format_exception(exc_type, exc_value, exc_traceback))
    # Display the error
    QMessageBox.critical(None, "Critical Error", f"An unhandled exception occurred:\n{exc_value}")


    if __name__ == "__main__":
    # Replace QApplication with CustomApplication
    app = QApplication(sys.argv)

    # Create main window
    main_window = MainWindow()
    main_window.show()


    # Set global exception hook
    sys.excepthook = exception_hook

    # Start the application event loop
    app.exec_()


    I need to find a solution to handle exceptions from a main thread, to be able to debug and log correctly the application

    I'm working on a conda env with python 3.6 on a windows machine

    Continue reading...

Compartilhe esta Página