当前位置：首页 > article >正文

PyQt6/PySide6 的 SQL 数据库操作（QtSql）

article 2025/2/24 9:23:39

一、核心组件架构

1.1 QtSql模块构成

QSqlDatabase：数据库连接管理（支持连接池）
QSqlQuery：SQL语句执行与结果遍历
QSqlTableModel：可编辑的表格数据模型
QSqlQueryModel：只读查询结果模型
QSqlRelationalTableModel：支持外键关系模型

1.2 数据库驱动支持

print(QSqlDatabase.drivers())  # 输出可用驱动列表
# 典型输出：['QSQLITE', 'QMYSQL', 'QPSQL', 'QODBC']

二、数据库连接实战

2.1 SQLite连接示例

def create_connection():
    db = QSqlDatabase.addDatabase('QSQLITE')
    db.setDatabaseName('contacts.db')
  
    if not db.open():
        QMessageBox.critical(None, "数据库错误",
            f"连接失败：{db.lastError().text()}")
        return False
  
    # 验证表存在性
    required_tables = {'contacts'}
    existing_tables = db.tables(QSql.Tables)
    if not required_tables.issubset(existing_tables):
        init_database(db)
  
    return True

def init_database(conn):
    query = QSqlQuery()
    query.exec("""
        CREATE TABLE contacts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            phone TEXT UNIQUE,
            email TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    # 创建索引
    query.exec("CREATE INDEX idx_contacts_name ON contacts(name)")

2.2 MySQL连接配置

db = QSqlDatabase.addDatabase('QMYSQL')
db.setHostName('localhost')
db.setPort(3306)
db.setDatabaseName('mydb')
db.setUserName('root')
db.setPassword('secret')
db.setConnectOptions("MYSQL_OPT_RECONNECT=1;")

三、CRUD操作进阶

3.1 参数化查询（防SQL注入）

# 插入数据
query = QSqlQuery()
query.prepare("""
    INSERT INTO contacts 
    (name, phone, email)
    VALUES (?, ?, ?)
""")
query.addBindValue("张三")
query.addBindValue("13800138000")
query.addBindValue("zhangsan@example.com")
query.exec()

# 批量插入
names = [("李四", "13912345678"), ("王五", "13687654321")]
query.prepare("INSERT INTO contacts (name, phone) VALUES (?, ?)")
for name, phone in names:
    query.addBindValue(name)
    query.addBindValue(phone)
    query.exec()

3.2 复杂查询与结果处理

query = QSqlQuery()
query.exec("""
    SELECT id, name, phone, 
           strftime('%Y-%m-%d', created_at) AS create_date
    FROM contacts
    WHERE name LIKE ?
    ORDER BY created_at DESC
    LIMIT 10
""")
query.addBindValue("%张%")

while query.next():
    record = query.record()
    print(f"ID: {query.value(0)}, Name: {query.value('name')}")
    print(f"Phone: {record.value(2)}, Date: {query.value(3)}")

四、模型-视图编程

4.1 QSqlTableModel实时同步

class ContactManager(QWidget):
    def __init__(self):
        self.model = QSqlTableModel()
        self.model.setTable("contacts")
        self.model.setEditStrategy(QSqlTableModel.OnFieldChange)
        self.model.select()

        self.view = QTableView()
        self.view.setModel(self.model)
        self.view.setColumnHidden(0, True)  # 隐藏ID列
        self.view.doubleClicked.connect(self.edit_contact)

    def add_contact(self):
        row = self.model.rowCount()
        self.model.insertRow(row)
        self.model.setData(self.model.index(row, 1), "新联系人")
        self.model.submitAll()

    def delete_contact(self):
        index = self.view.currentIndex()
        if index.isValid():
            self.model.removeRow(index.row())
            self.model.submitAll()

4.2 自定义查询模型

class FilterProxyModel(QSortFilterProxyModel):
    def __init__(self, pattern, parent=None):
        super().__init__(parent)
        self.pattern = QRegularExpression(pattern, 
            QRegularExpression.CaseInsensitiveOption)

    def filterAcceptsRow(self, row, parent):
        model = self.sourceModel()
        index = model.index(row, 1)  # 过滤name列
        return self.pattern.match(model.data(index)).hasMatch()

# 使用示例
custom_model = QSqlQueryModel()
custom_model.setQuery("SELECT * FROM contacts")
proxy = FilterProxyModel("张")
proxy.setSourceModel(custom_model)
view.setModel(proxy)

五、事务与性能优化

5.1 事务处理模板

db = QSqlDatabase.database()
db.transaction()

try:
    # 执行多个操作
    insert_query.exec()
    update_query.exec()
    db.commit()
except:
    db.rollback()
    raise

5.2 连接池配置

from PySide6.QtCore import QSettings

def setup_connection_pool():
    settings = QSettings("config.ini", QSettings.IniFormat)
  
    QSqlDatabase.setConnectOptions(
        f"QSQLITE_MAX_CONNECTIONS=20;"
        f"QSQLITE_BUSY_TIMEOUT=5000;"
    )
  
    for i in range(5):  # 初始化5个连接
        conn = QSqlDatabase.addDatabase('QSQLITE', f"conn_{i}")
        conn.setDatabaseName(settings.value("Database/Path"))

六、实战案例：通讯录管理系统

功能清单：

联系人增删改查
拼音快速检索
导出CSV功能
数据库备份/恢复
批量导入Excel

关键代码实现：

class ContactBook(QMainWindow):
    def export_csv(self):
        with open('contacts.csv', 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['姓名', '电话', '邮箱'])
          
            query = QSqlQuery("SELECT name, phone, email FROM contacts")
            while query.next():
                writer.writerow([
                    query.value(0),
                    query.value(1),
                    query.value(2)
                ])

七、常见问题解决方案

7.1 中文乱码处理

# MySQL连接字符串添加charset参数
db.setConnectOptions("MYSQL_OPT_SET_CHARSET_NAME=UTF8MB4;")

# SQLite设置编码
query.exec("PRAGMA encoding = 'UTF-8'")

7.2 数据库迁移方案

使用Alembic进行版本控制：

# 版本迁移脚本示例
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table('contacts',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String(50), nullable=False)
    )

def downgrade():
    op.drop_table('contacts')