PyQt6结合FastAPI/UVicorn调用QTextEdit.setText()崩溃求助

使用PyQt6.4.0 + FastAPI0.89.1 + Uvicorn0.20.0搭建带API服务的桌面程序，为避免界面卡顿将FastAPI服务放在子线程运行，但调用API尝试更新QTextEdit文本时程序崩溃，错误码-1073741819 (0xC0000005)；移除修改UI的代码后，API可正常响应。

Qt的所有UI组件（如QTextEdit）只能在主线程（UI线程）中访问和修改，FastAPI服务运行在子线程，直接在子线程的API接口里调用self.textedit.setText()属于跨线程操作UI，触发内存访问违规，导致程序崩溃。

以下三种方案均可实现通过API更新UI，且符合Qt线程安全规范：

方案一：使用Qt信号与槽（推荐，遵循Qt设计规范） #

通过定义信号，将子线程的UI更新请求通知主线程，由主线程的槽函数完成UI修改：

from fastapi import FastAPI
import threading
import uvicorn
from PyQt6.QtWidgets import (QMainWindow, QApplication, QTextEdit)
import sys
from PyQt6.QtCore import (QRect, pyqtSignal)


class MainWindow(QMainWindow):
    # 定义更新UI的信号
    update_text_signal = pyqtSignal(str)

    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
        self.setGeometry(QRect(0, 0, 400, 400))
        self._initial_widgets()
        # 绑定信号与槽函数
        self.update_text_signal.connect(self._update_textedit)
        self._create_server()

    def _initial_widgets(self):
        self.textedit = QTextEdit(self)
        self.textedit.setGeometry(QRect(100, 0, 100, 100))
        self.textedit.setReadOnly(True)
        self.setCentralWidget(self.textedit)

    def _update_textedit(self, text):
        # 槽函数在主线程执行，安全修改UI
        self.textedit.setText(text)

    def _create_server(self):
        self.app = FastAPI()

        @self.app.post("/")
        async def change_textEdit_data():
            # 发射信号，通知主线程更新UI
            self.update_text_signal.emit("hello")
            return {"detail": "OK"}

        thread = threading.Thread(target=uvicorn.run, kwargs={"app": self.app, "port": 80})
        thread.daemon = True  # 设置守护线程，关闭主程序时自动退出子线程
        thread.start()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    app.exec()

方案二：使用QMetaObject.invokeMethod（轻量方案） #

直接通过QMetaObject.invokeMethod强制在主线程执行UI修改操作，无需定义信号槽：

from fastapi import FastAPI
import threading
import uvicorn
from PyQt6.QtWidgets import (QMainWindow, QApplication, QTextEdit)
import sys
from PyQt6.QtCore import (QRect, QMetaObject, Qt)


class MainWindow(QMainWindow):
    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
        self.setGeometry(QRect(0, 0, 400, 400))
        self._initial_widgets()
        self._create_server()

    def _initial_widgets(self):
        self.textedit = QTextEdit(self)
        self.textedit.setGeometry(QRect(100, 0, 100, 100))
        self.textedit.setReadOnly(True)
        self.setCentralWidget(self.textedit)

    def _create_server(self):
        self.app = FastAPI()

        @self.app.post("/")
        async def change_textEdit_data():
            # 强制在主线程执行setText操作
            QMetaObject.invokeMethod(self.textedit, "setText",
                                     Qt.ConnectionType.QueuedConnection,
                                     Qt.Q_ARG(str, "hello"))
            return {"detail": "OK"}

        thread = threading.Thread(target=uvicorn.run, kwargs={"app": self.app, "port": 80})
        thread.daemon = True
        thread.start()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    app.exec()

方案三：替换Uvicorn为Qt网络模块（全Qt生态方案） #

如果不想依赖FastAPI和Uvicorn，可使用Qt自带的QTcpServer实现简单的API服务，完全遵循Qt线程模型：

from PyQt6.QtWidgets import (QMainWindow, QApplication, QTextEdit)
from PyQt6.QtCore import (QRect, QTcpServer, QTcpSocket)
import sys


class MainWindow(QMainWindow):
    def __init__(self, *args, **kwargs):
        super(MainWindow, self).__init__(*args, **kwargs)
        self.setGeometry(QRect(0, 0, 400, 400))
        self._initial_widgets()
        self._start_tcp_server()

    def _initial_widgets(self):
        self.textedit = QTextEdit(self)
        self.textedit.setGeometry(QRect(100, 0, 100, 100))
        self.textedit.setReadOnly(True)
        self.setCentralWidget(self.textedit)

    def _start_tcp_server(self):
        self.server = QTcpServer(self)
        # 监听80端口
        if not self.server.listen(port=80):
            print("Server start failed")
            return
        # 新连接触发信号
        self.server.newConnection.connect(self._handle_new_connection)

    def _handle_new_connection(self):
        socket = self.server.nextPendingConnection()
        socket.readyRead.connect(lambda: self._read_socket_data(socket))
        socket.disconnected.connect(socket.deleteLater)

    def _read_socket_data(self, socket):
        # 读取请求数据
        data = socket.readAll().data().decode()
        # 简单判断POST请求
        if "POST / HTTP/1.1" in data:
            # 主线程直接修改UI
            self.textedit.setText("hello")
            # 返回响应
            response = b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n{\"detail\": \"OK\"}"
            socket.write(response)
        socket.disconnectFromHost()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    app.exec()

• 方案一和方案二保留了FastAPI的功能，适合需要RESTful API的场景；
• 方案三完全基于Qt，无需额外依赖，适合轻量服务场景；
• 所有方案均保证UI操作在主线程执行，避免线程安全问题。

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