Qt小组件 - 7 SQL Thread Qt访问数据库ORM

简介

网上关于Qt访问数据库的资料大多使用QSqlDatabase模块。虽然这在C++中尚可接受，但在Python中使用就显得过于繁琐了——不仅要手动编写SQL语句，还与Python追求简洁的理念背道而驰。

在这里写一个基于sqlalchemy的示例，也可以使用其他的ORM库

sql_thread.py

# coding: utf-8
from collections import deque
from typing import Callable

from PySide6.QtCore import QObject, Signal, QThread
from sqlalchemy.orm import Session


class SqlRequest:
    """ Sql request """

    def __init__(self, func: Callable, slot: Callable = None, params: dict = None):
        self.func = func
        self.slot = slot
        self.params = params or {}


class SqlResponse:
    """ Sql的反应 """

    def __init__(self, data: object, slot: Callable = None):
        self.slot = slot
        self.data = data


class SqlSignalBus(QObject):
    """ Sql信号总线 """

    fetchDataSig = Signal(SqlRequest)
    dataFetched = Signal(SqlResponse)


sqlSignalBus = SqlSignalBus()


def sqlRequest(func: Callable, slot=None, **params):
    """ 从数据库查询 SQL 数据 """
    request = SqlRequest(func, slot, params)
    sqlSignalBus.fetchDataSig.emit(request)


class DatabaseThread(QThread):

    def __init__(self, parent=None):
        super().__init__(parent=parent)
        self.server = None
        self.tasks = deque()

        sqlSignalBus.fetchDataSig.connect(self.onFetchData)

    def run(self):
        func, args, kwargs = self.server
        server: Session = func(*args, **kwargs)
        while self.tasks:
            request: SqlRequest = self.tasks.popleft()
            result = request.func(**request.params, server=server)
            sqlSignalBus.dataFetched.emit(SqlResponse(result, request.slot))
        server.close()

    def setServer(self, func: Callable, *args, **kwargs):
        self.server = (func, args, kwargs)

    def onFetchData(self, request: SqlRequest):
        self.tasks.append(request)

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

db.py

# coding: utf-8
from random import randint, choice

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

engine = create_engine('sqlite:///test.db')
Base = declarative_base()


def create_session():
    return sessionmaker(bind=engine)()


def init_create_table() -> None:
    Base.metadata.create_all(bind=engine, checkfirst=True)


class User(Base):
    # 表的名字:
    __tablename__ = 'user'

    # 表的结构:
    id = Column(Integer, autoincrement=True, primary_key=True, comment='用户ID')
    name = Column(String(20))
    age = Column(Integer)
    phone = Column(String(20))
    email = Column(String(50))
    sex = Column(String(10))

    def __repr__(self):
        return f"User(id={self.id}, name={self.name}, age={self.age}, phone={self.phone}, email={self.email}, sex={self.sex})"


def add_user(server: Session):
    server.add(
        User(name=f'Tom-{randint(1, 100)}', age=randint(1, 100), phone=str(randint(10000000, 99999999)),
             email='tom@123.com',
             sex=choice(['male', 'female'])))
    server.commit()


def select_user(server: Session):
    users = server.query(User).all()
    return users

demo.py

# coding: utf-8
import sys

from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QHBoxLayout

from db import create_session, User, add_user, select_user, init_create_table
from sql_thread import sqlSignalBus, SqlResponse, DatabaseThread, sqlRequest


class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Sql Thread Example")
        self.horizontalLayout = QHBoxLayout(self)
        self.addUserButton = QPushButton("Add User", self)
        self.selectUserButton = QPushButton("Select User", self)
        self.horizontalLayout.addWidget(self.addUserButton)
        self.horizontalLayout.addWidget(self.selectUserButton)

        self.resize(400, 300)

        self.databaseThread = DatabaseThread(self)
        self.databaseThread.setServer(create_session)
        sqlSignalBus.dataFetched.connect(self.onDataFetched)

        self.addUserButton.clicked.connect(lambda: sqlRequest(add_user))
        self.selectUserButton.clicked.connect(lambda: sqlRequest(select_user, self.onSelectUser))

    def onSelectUser(self, users: list[User]):
        print(users)

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


if __name__ == '__main__':
    app = QApplication(sys.argv)
    init_create_table()
    mainWindow = MainWindow()
    mainWindow.show()
    sys.exit(app.exec())