簡介

網上關于Qt訪問數據庫的資料大多使用QSqlDatabase模塊。雖然這在C++中尚可接受，但在Python中使用就顯得過于繁瑣了——不僅要手動編寫SQL語句，還與Python追求簡潔的理念背道而馳。

在這里寫一個基于sqlalchemy的示例，也可以使用其他的ORM庫

sql_thread.py

# coding: utf-8
from collections import deque
from typing import Callablefrom PySide6.QtCore import QObject, Signal, QThread
from sqlalchemy.orm import Sessionclass SqlRequest:""" Sql request """def __init__(self, func: Callable, slot: Callable = None, params: dict = None):self.func = funcself.slot = slotself.params = params or {}class SqlResponse:""" Sql的反應 """def __init__(self, data: object, slot: Callable = None):self.slot = slotself.data = dataclass 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 = Noneself.tasks = deque()sqlSignalBus.fetchDataSig.connect(self.onFetchData)def run(self):func, args, kwargs = self.serverserver: 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, choicefrom sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Sessionengine = 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 sysfrom PySide6.QtWidgets import QApplication, QWidget, QPushButton, QHBoxLayoutfrom db import create_session, User, add_user, select_user, init_create_table
from sql_thread import sqlSignalBus, SqlResponse, DatabaseThread, sqlRequestclass 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())