PyQt实现异步数据库请求的实战记录

def sqlRequest(
    service: str, 
    method: str, 
    slot, 
    params: dict = None
)

class SignalBus(QObject):
    """ Signal bus """
    fetchDataSig = Signal(SqlRequest)    # 请求数据信号
    dataFetched = Signal(SqlResponse)    # 响应数据信号

    
signalBus = SignalBus()
    
    
class SqlRequest:
    """ Sql request """

    def __init__(self, service: str, method: str, slot=None, params: dict = None):
        self.service = service
        self.method = method
        self.slot = slot
        self.params = params or {}


class SqlResponse:
    """ Sql response """

    def __init__(self, data, slot):
        self.slot = slot
        self.data = data

def sqlRequest(service: str, method: str, slot=None, params: dict = None):
    """ query sql from database """
    request = SqlRequest(service, method, slot, params)
    signalBus.fetchDataSig.emit(request)

from qfluentwidgets import HeaderCardWidget, ComboBox

class LicenseCard(HeaderCardWidget):
    
    def __init__(self, parent=None):
        super().__init__("许可证", parent)
        self.goodsComboBox = ComboBox(self)
        
        # 请求商品信息
        sqlRequest("goodsService", "listAll", self.onGoodsFetched)

    def onGoodsFetched(self, goods: List[Goods]):
        """ 将商品信息添加到下拉框中 """
        for good in goods:
            self.goodsComboBox.addItem(good.name, userData=good)

class DatabaseThread(QThread):
    """ Database thread """

    def __init__(self, db: QSqlDatabase = None, parent=None):
        super().__init__(parent=parent)
        self.database = Database(db, self)
        self.tasks = deque()

        # 处理请求信号
        signalBus.fetchDataSig.connect(self.onFetchData)

    def run(self):
        """ 处理请求 """
        while self.tasks:
            task, request = self.tasks.popleft()
            result = task(**request.params)
            signalBus.dataFetched.emit(SqlResponse(result, request.slot))

    def onFetchData(self, request: SqlRequest):
        """ 将请求添加到队列中 """
        service = getattr(self.database, request.service)
        task = getattr(service, request.method)
        self.tasks.append((task, request))

        if not self.isRunning():
            self.start()
                

class Database(QObject):
    """ Database """

    def __init__(self, db: QSqlDatabase = None, parent=None):
        super().__init__(parent=parent)
        self.orderService = OrderService(db)
        self.userService = UserService(db)
        self.goodsService = GoodsService(db)

from qfluentwidgets import MSFluentWindow

class MainWindow(MSFluentWindow):
    """ 主界面 """
    
    def __init__(self):
        super().__init__()
        
        # 处理响应结果
        signalBus.dataFetched.connect(self.onDataFetched)

    def onDataFetched(self, response: SqlResponse):
        if response.slot:
            response.slot(response.data)