简介:基于Python PyQt5开发的数据库操作小工具,包含完整源码与SQLite数据库,适合Python初学者、GUI编程爱好者以及有轻量级数据管理需求的开发者。该工具以PyQt5搭建界面,支持数据库连接、查询、插入、更新、删除等常用操作,并通过具体示例展示异常处理与事务控制方法,是学习桌面应用与数据库交互的实用案例。压缩包大小约12.48MB,共171个文件,其中py脚本承载主逻辑,ui文件定义窗口布局,bmp为界面图标资源,cpp/h及pro文件与工程编译相关,另含db3数据库文件,可直接运行体验;目录结构清晰,便于按模块查找。目前已有119人下载学习。研读源码可掌握PyQt5的信号槽连接、QTableWidget等控件使用,以及sqlite3模块从连接、游标操作到结果集处理的完整流程;封装的DatabaseManager类还提供了面向对象的分层思路,可在此基础上快速扩展为更完整的数据库管理系统。
1. 一个PyQt5数据库小工具,解决的是谁每天重复的劳动
我见过不少同事,临时想确认库里一张表的数据情况,得先打开重型客户端,输一堆连接信息,再点连接、找表、看数据;写一条调试 SQL,要么在命令行里被结果糊满屏,要么在客户端里等半天。基于 Python PyQt5 实现的数据库操作小工具,就是把「连接数据库—看表结构—写 SQL—查结果—导出」这一套动作压缩到一个桌面窗口里。它不试图替代专业客户端,而是让查数、改数据、导结果变成几秒钟的事。标题里的“源码+数据库”意味着你拿到的是一套能直接跑的工程,附带一个示例数据库,省去自己造数的时间。适合想自己掌控工具、愿意在现有源码上做二次开发的 Python 开发者、测试、数据分析和运维同学。
2. 先拆结构:PyQt5界面与数据库操作怎么分层,才不会越改越乱
小工具写到第三天,最容易碰到的是这个翻车现场:在按钮回调里直接写 SQL,每点一次查询界面就卡住了;数据准备好了,又不知道该往哪个控件里填;界面元素一多,改一个效果要连带改三处。根源不是写得少,而是没分层。
2.1 界面刷新慢、数据错乱,多半是分层出了问题
PyQt5 本质上是 GUI 框架,它管的是窗口、控件、事件循环;数据库驱动管的是连接、SQL、结果集。把 SQL 丢在按钮回调里执行,界面线程就得原地等数据库返回,十几万行的表查下来,窗口直接变成“未响应”。更隐蔽的问题在子线程:有人在 QThread 里直接调用self.table_result.setItem(),但 Qt 的控件不是线程安全的,轻则闪烁错乱,重则崩溃。
正确做法是让界面层只发请求、收结果,数据层只跟 SQL 打交道,中间用信号槽传数据。界面层不碰数据库驱动,数据层不碰控件。这样做的直接收益是:以后你想把 SQLite 换成 MySQL,只需要动数据层,界面按钮一行不用改。
2.2 三个模块的职责边界与调用方向
小工具一般拆三层就够了,我给一个常用切分:
| 模块 | 负责 | 明确不做什么 |
|---|---|---|
| 界面层 MainWindow | 接收用户操作、显示结果、状态栏反馈 | 不直接执行 SQL,不持有连接 |
| 适配层 TableService(可选) | 把数据层返回的 dict 列表整理成界面所需结构 | 不关心按钮和布局 |
| 数据层 DBManager | 连接、参数化 SQL、事务提交回滚、释放连接 | 不 import 任何 QWidget |
调用方向是单向的:MainWindow → TableService → DBManager → 数据库。禁止反向。为什么反复强调这个?因为一旦界面层直接import sqlite3,后面想换数据库就得在界面代码里到处找 SQL 字符串。我的习惯是让界面层只认识execute_any、fetch_all这类方法名,数据层负责屏蔽 SQLite 和 MySQL 的差异。
适配层不是必须的。表不超过五张、查询不超过十种时,硬加一层反而绕。我一般是等到清理动作变多,再把「查用户列表」「按 ID 删记录」这类操作收敛成具名方法,按钮回调里就不会出现裸 SQL 字符串。
2.3 数据库连接用单例还是每次新建:我为什么选短连接
桌面小工具最常见的做法是全局建一个连接对象,谁都能拿过来用,结构简单,但后患不小。SQLite 的长连接容易把写锁占住,另一个表再写就报database is locked;MySQL 长连接空闲超过服务端 wait_timeout 会直接断开,第二天再点就报连接丢失。所以我的做法是短连接:每次执行 SQL 时新建连接,用完立即关闭。
import sqlite3 class DBManager: """数据层:只负责连接与 SQL 执行,不 import 任何界面控件""" def __init__(self, db_path, timeout=10): self.db_path = db_path self.timeout = timeout def _connect(self): # 每次执行都新建连接,避免长连接占用锁 conn = sqlite3.connect(self.db_path, timeout=self.timeout) conn.row_factory = sqlite3.Row # 结果集支持按列名访问 conn.execute("PRAGMA foreign_keys = ON") return conn def fetch_all(self, sql, params=None): conn = self._connect() try: cur = conn.execute(sql, params or ()) rows = [dict(row) for row in cur.fetchall()] columns = [col[0] for col in cur.description] if cur.description else [] return columns, rows finally: conn.close()这段代码有三个值得注意的参数点。timeout=10是 SQLite 等待锁的秒数,默认只有 5 秒,设到 10 是给并发写留余量。row_factory = sqlite3.Row让每行能通过列名取值,转dict(row)后信号槽传起来也方便。PRAGMA foreign_keys = ON比较容易被忽略,SQLite 默认不开外键约束,每个连接都要单独开启。
短连接的代价是每次执行多一次文件打开和握手,但对桌面工具这种低频操作完全可以接受。后来我接 MySQL 时也是同理,每次连一次,查完关掉,基本没有再遇到过连接断掉的问题。
3. 从源码跑起来:环境安装、目录结构与最小启动命令
拿到源码后第一件事不是改代码,而是先在本地把它跑起来。这一步顺了,后面改什么都心里有底。
3.1 环境准备:Python 版本与依赖清单
这个方案依赖非常简单,SQLite 部分不需要额外驱动,Python 自带sqlite3标准库;界面部分只需 PyQt5。如果你的目标数据库是 MySQL,再补一个 PyMySQL。我习惯把依赖写进 requirements.txt:
PyQt5>=5.15 PyMySQL>=1.0然后用 venv 隔离环境,避免把系统 Python 搞乱:
python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate pip install -r requirements.txt python main.py先建虚拟环境再装依赖,是这类桌面小工具最不容易踩坑的启动方式。PyQt5 的安装包比较大,pip 装的时候稍微等一会儿是正常的。值得注意的是,如果源码里用了 Qt Designer 画的.ui文件,需要先执行pyuic5 main_window.ui -o main_window_ui.py把界面定义转成 Python 模块,再运行 main.py。
3.2 源码目录长什么样:入口、界面、数据库文件
拿到源码后我一般最先看目录结构,判断这个工程是不是“能跑”的状态。典型的布局长这样:
db_tool/ ├── main.py # 应用入口:创建 QApplication ├── db_manager.py # 数据层:连接、SQL、事务 ├── main_window.py # 界面层:主窗口布局与信号槽 ├── config.ini # 连接参数配置 ├── requirements.txt ├── ui/ │ └── main_window.ui # Qt Designer 绘制的界面,可选 └── data/ └── demo.db # 附带的 SQLite 示例数据库标题里的“源码 + 数据库”指的就是这套结构里既有程序代码,也带了可连接的数据库文件。示例库里一般会放两张演示表,比如一张用户表、一张订单表,里面预置几条能验证增删改查的数据。这样你打开工具后不用先造数,直接就能看到左侧树和结果表格都有内容。
如果看到源码里只有.ui文件、没有对应的_ui.py文件,说明作者通常是在运行前用 pyuic5 生成的。这种工程跑不起来多半是漏了这一步,不是代码本身有问题。
3.3 最小启动命令与首次连接验证
main.py 是这个工具的入口,职责非常简单:创建应用、创建窗口、进入事件循环。给它足够的瘦身,以后维护时才不需要在入口文件里翻业务逻辑。
import sys from PyQt5.QtWidgets import QApplication from main_window import MainWindow def main(): app = QApplication(sys.argv) win = MainWindow("config.ini") win.show() sys.exit(app.exec_()) if __name__ == "__main__": main()启动后建议按三步验证是否正常:先看状态栏是否显示“已连接 data/demo.db”,这证明配置读取和连接建立没问题;再看左侧树是否列出了 demo_user、demo_order 两张表,这证明元数据查询成功;最后在 SQL 编辑器里输入select * from demo_user;,按执行快捷键,下方表格出现行数据,整个链路就通了。
这套验证顺序是沿着依赖方向走的:先验证底层连接,再验证界面到数据的通路,最后验证完整链路。如果第二步没表,问题大概率在连接参数;如果第三步没结果,问题在 SQL 执行分支。
3.4 把连接参数抽到 config.ini,避免写死在代码里
写死连接串是这类小工具最容易后悔的地方。今天连的是 demo.db,明天要连测试库就得改代码重跑。我一般把连接参数全抽到 config.ini:
[database] type = sqlite path = data/demo.db timeout = 10 ; 如果接 MySQL,改成下面这种配置 ; type = mysql ; host = 127.0.0.1 ; port = 3306 ; user = root ; password = 123456 ; schema = demo_db读取配置的代码同样放在数据层,避免界面层直接接触配置文件:
import configparser def load_config(config_path): cfg = configparser.ConfigParser() cfg.read(config_path, encoding="utf-8") return cfg["database"]configparser 读 ini 文件时,encoding="utf-8"要显式声明,否则在 Windows 默认编码环境下,配置里的中文注释可能直接报解码错误。type字段用来决定创建哪种连接,常见写法是在一个工厂函数里判断:
def build_db(cfg): db_type = cfg.get("type", "sqlite") if db_type == "sqlite": return DBManager(db_path=cfg.get("path", "data/demo.db"), timeout=float(cfg.get("timeout", "10"))) if db_type == "mysql": # 用 PyMySQL 实现的 DBManager 分支,接口保持一致 return MySQLDBManager(cfg) raise ValueError(f"暂不支持的数据库类型: {db_type}")这样做的好处是界面层完全不用关心你连的是什么库。今天用 SQLite 跑通,明天切换 MySQL 只需要改配置和换数据层实现,主窗口代码一行不动。
4. 把数据库操作做成“看得见”的样子:表树、SQL编辑器与结果表格
工具的核心价值在于能同时看到“库里面有什么”和“查出来是什么”。这一章把主窗口的骨架、表结构加载、SQL 执行和参数化封装一次说清楚。
4.1 主窗口布局:为什么用“左树、右编辑器、下表格”
我一般用 QSplitter 搭三块区域:左侧 QTreeWidget 显示表 / 视图,右上 QPlainTextEdit 写 SQL,右下 QTableWidget 展示结果。整体思路是让常用的操作都在一个窗口里完成,不弹额外的子窗口。
from PyQt5.QtWidgets import ( QMainWindow, QWidget, QSplitter, QTreeWidget, QPlainTextEdit, QTableWidget, QVBoxLayout ) class MainWindow(QMainWindow): def __init__(self, db): super().__init__() self.db = db self.setWindowTitle("数据库操作小工具") self._build_ui() def _build_ui(self): splitter = QSplitter() self.tree_tables = QTreeWidget() self.editor_sql = QPlainTextEdit() self.table_result = QTableWidget() right = QWidget() right_box = QVBoxLayout(right) right_box.addWidget(self.editor_sql, 3) # 伸缩系数 3 right_box.addWidget(self.table_result, 4) # 伸缩系数 4 right_box.setContentsMargins(0, 0, 0, 0) splitter.addWidget(self.tree_tables) splitter.addWidget(right) splitter.setStretchFactor(0, 1) splitter.setStretchFactor(1, 3) self.setCentralWidget(splitter)QSplitter 的伸缩系数决定窗口拉大时各区域按什么比例分配空间。编辑器 3、结果表格 4,意思是在新增空间里表格多分一点,因为查出来的数据比 SQL 文本更占视野。QPlainTextEdit 而不是 QTextEdit,是为了避免富文本格式干扰,SQL 编辑器只需要纯文本。
结果展示这里,几千行以内 QTableWidget 够用,它按单元格填数据,直观好调试;但如果要查几十万行,QTableWidget 会明显变慢,那种场景要换 QTableView 加 QStandardItemModel 做虚拟滚动。小工具的定位决定了我不提前上重型表格模型。
4.2 读取表结构和数据:元数据查询与树填充
SQLite 的表清单存在系统表sqlite_master里,这是加载左侧树的最直接来源:
SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name;过滤sqlite_%是为了把 sqlite_sequence 这类系统表藏起来,只展示用户表。填充树的代码如下:
def refresh_tables(self): columns, rows = self.db.fetch_all( "SELECT name FROM sqlite_master " "WHERE type='table' AND name NOT LIKE 'sqlite_%' " "ORDER BY name" ) self.tree_tables.clear() for row in rows: from PyQt5.QtWidgets import QTreeWidgetItem self.tree_tables.addTopLevelItem(QTreeWidgetItem([row["name"]]))点击树节点时再查字段结构,用 PRAGMA 命令:
PRAGMA table_info('demo_user');返回的每一行里包含cid、name、type、notnull、dflt_value、pk几个字段。我们可以取name和type拼成“字段名 类型”展示,也可以直接放到一个只读表格里给用户看。
表名传进 PRAGMA 时需要注意单引号转义,万一表名里带了单引号,把'替换成''即可。表名来源如果是 sqlite_master 查出来的,可信度较高;如果是用户手动输入的文本,则要做白名单校验,避免拼接语句出问题。
4.3 SQL执行与增删改查:一条通用执行入口
工具里最核心的方法是“不管查什么,先执行了再判断”。判断依据是游标的description属性:有列描述说明是查询语句,需要取数据;没有列描述说明是写语句,需要提交事务。
def execute_any(self, sql): """通用执行入口:根据 description 自动区分查询与写操作""" conn = self._connect() try: cur = conn.execute(sql) if cur.description: rows = [dict(row) for row in cur.fetchall()] return "query", cur.description, rows else: conn.commit() return "update", cur.rowcount, None finally: conn.close()用description判断比“SQL 开头是不是 select”更可靠。with开头的 CTE 查询也能返回列描述,注释开头的 SQL 也不会被误判。这个方案还能直接兼容 MySQL,因为 PyMySQL 的游标同样有 description 属性。
调用方只需要关心返回的三种情况:查询时拿到列描述和行数据,更新时拿到影响行数。主窗口里填表格的逻辑可以收敛成一个fill_table(columns, rows)方法,每次查询结果回来先清空旧数据再填充新表头。
这里有个边界要提醒:execute_any一次只执行一条 SQL。如果你往编辑器里贴了三条 INSERT,中间再带个分号,执行到第二条可能报语法错误。常见做法是在执行前提示用户“一次只执行一条语句”,或者自己按分号拆分,但分号拆分在遇到字符串里的分号时会误切,我并不推荐。
4.4 参数化查询封装:别再拿 f-string 拼 SQL
工具做成后,很多人会顺手加几个快捷按钮,比如“按 ID 删除记录”。这时候最容易写出的代码就是f"DELETE FROM demo_user WHERE id = {user_id}",看着简单,实际上是在给 SQL 注入开口子。
def delete_user_by_id(self, user_id): sql = "DELETE FROM demo_user WHERE id = ?" return self.db.execute_update(sql, (user_id,))socket库的?占位符会把参数原样交给驱动转义,不会跟 SQL 字符串混在一起。如果接到 MySQL,占位符要换成%s,其他逻辑不变。
批量写入场景用executemany更合适:
def batch_insert_users(self, items): conn = self._connect() try: conn.executemany( "INSERT INTO demo_user(name, age) VALUES(?, ?)", [(item["name"], item["age"]) for item in items] ) conn.commit() finally: conn.close()executemany接收一个可迭代的元组列表,驱动会自动复用同一条语句反复执行,比逐条 execute 快不少。数据量大时可以分批传入,比如每次 500 条,避免一次事务太长。
5. 血泪避坑:Qt线程卡界面、中文乱码、SQLite锁库
这个工具能跑通和“能顺手用”之间,隔着几个大概率踩中的坑。下面每一条都是我实际改过的,按“现象 → 原因 → 解决”记下来,你可以直接对着排查。
5.1 大查询把界面卡成“未响应”
现象:查询一张几十万行的表,窗口标题栏出现“无响应”,鼠标转圈,哪怕结果集只有几百行,只要查询本身复杂,界面就卡住了。
原因:SQL 在 GUI 线程里同步执行。界面线程就是执行查询的线程,查询期间它无法处理重绘、鼠标事件,操作系统判断程序未响应。PyQt5 里的所有控件操作都必须在主线程,这是前提,但这并不意味着你必须在主线程执行 SQL。
解决:把耗时查询放进 QThread,查询完用信号把结果发回主线程。需要注意的是,sqlite3 连接默认不允许跨线程使用,所以子线程里执行查询时,连接必须在子线程内新建。我那版 DBManager 每次执行都新建连接,恰好保证了这一点。
from PyQt5.QtCore import QThread, pyqtSignal class QueryThread(QThread): result_ready = pyqtSignal(list, list) # columns, rows def __init__(self, db, sql): super().__init__() self.db = db self.sql = sql def run(self): columns, rows = self.db.fetch_all(self.sql) self.result_ready.emit(columns, rows)启动线程时注意保存引用,避免线程对象被垃圾回收:
def run_query_in_thread(self, sql): self.query_thread = QueryThread(self.db, sql) self.query_thread.result_ready.connect(self.fill_table) self.query_thread.finished.connect(self.query_thread.deleteLater) self.query_thread.start()finished信号连到deleteLater是常用的清理套路,线程跑完自动销毁自己,不拖到窗口关闭才回收。如果你想加“取消查询”按钮,不要用terminate(),那是强行结束线程,SQLite 连接可能处于中间状态。更稳的做法是在线程类里加一个请求标志,循环里定期检查。
5.2 中文显示成乱码
现象:SQLite 里存的中文显示正常,但从外部文件读入 SQL、或者接 MySQL 后,中文变成??或一团乱码。
原因:这类问题通常不是 PyQt5 造成的,而是三个环节编码不一致:Python 读取文件时的编码、数据库连接字符集、终端输出环境。PyQt5 控件本身是 Unicode,不乱码。
解决:全部统一到 UTF-8。Python 源码默认 UTF-8 一般不用管;读取 SQL 文件时显式声明:
with open("query.sql", encoding="utf-8") as f: sql = f.read()接 MySQL 时,在连接参数里显式指定字符集:
import pymysql conn = pymysql.connect( host=host, port=port, user=user, password=pwd, database=schema, charset="utf8mb4", )utf8mb4比utf8更完整,能存表情符号。Windows 控制台打印中文乱码又是另一个问题,把命令行切到 UTF-8 代码页chcp 65001就好,但这只影响 print,不影响 GUI 表格显示。
5.3 SQLite 提示 database is locked
现象:执行 INSERT 或 UPDATE 时,程序抛出sqlite3.OperationalError: database is locked,重试几次还是报错。
原因:另一个连接持有写锁并且没有提交;或者同一个连接被多个线程同时使用;或者长连接里事务没结束,把锁一直占着。
解决:三个措施一起上,基本能消除这个报错。第一,连接时把 timeout 调到 10 秒,等待更久;第二,开启 WAL 模式,让读写并发互补阻塞;第三,代码里保证“每个线程独立连接,用后关闭”。
PRAGMA journal_mode=WAL; PRAGMA busy_timeout=3000;这三个参数的作用差异值得记一下:
| 参数 | 作用 | 建议值 |
|---|---|---|
| timeout | connect 时等待锁的秒数 | 10~30 |
| busy_timeout | 单条语句等待锁的毫秒数 | 3000 |
| journal_mode | 日志模式,WAL 提高读写并发 | WAL |
也不要以为开了 WAL 就不会锁库。WAL 只是让读不阻塞写、写不阻塞读,两个写者同时写照样要排锁。真正的教训是:写队列里只要有一条 SQL 没提交,后续写操作就会排队,排太久就报错。所以写操作尽量短平快,事务里不要夹带太多查询。
5.4 关闭窗口后进程不退出
现象:主窗口点 X 关掉了,shell 里命令一直不返回,进程还挂着。
原因:最常见的是 QThread 还在运行,PyQt5 关了窗口但子线程没有退出信号,进程被非守护线程拽住不放。有时候是因为代码里把closeEvent拦截后又ignore(),窗口看起来关掉了,实际还活着。
解决:在 closeEvent 里做退出前清理,先等线程收尾,再关数据库连接。
def closeEvent(self, event): if hasattr(self, "query_thread") and self.query_thread.isRunning(): self.query_thread.wait(2000) # 最多等 2 秒 if hasattr(self, "db"): self.db.close() event.accept()wait(2000)的意思是给线程两秒钟收尾,时间到了还没结束就继续关闭流程。不要在 wait 之后再跟线程有任何交互,因为可能已经结束了。真正要做到“随时可退”,最好在查询线程执行前加一条SELECT 1探活连接,不让子线程长时间阻塞在网络请求上。
5.5 跨线程传结果:信号里别传自定义对象
现象:把查询结果封装成一个自定义类,用emit(result_obj)发回主线程,槽函数里收到的对象结构异常,或者报出与内存地址相关的错误。
原因:跨线程信号传自定义类型,Qt 元对象系统需要知道如何拷贝和注册这个类型。没注册时,信号携带的只是一个不完整的引用,槽函数拿到的东西不是你想要的。
解决:信号参数只用基础类型,比如list、dict、str、int。我一般用pyqtSignal(list, list)传列名列表和行字典列表,简单直接。不要在信号里传数据库连接对象或游标,那东西跨线程本身就是不安全的。
6. 让工具变得更顺手:查询历史、导出CSV与启动自检
工具能跑通之后,真正决定你会不会每天打开它的是几个小细节。
查询历史是第一个值得加的。我在执行按钮的快捷键Ctrl+Enter回调里,先把编辑器里的 SQL 存进一个列表,再执行;同时给一个Ctrl+H快捷键轮询最近一条,按一下上一条就回来了。这个功能特别适合调试场景,手滑改坏一条复杂 SQL,还能找回来。
导出当前结果集是我建议第二个做的。查找数据之后总要把结果发给别人,我常用 CSV 导出,但注意 Windows 上 Excel 打开 CSV 中文会乱码,要用带 BOM 的 UTF-8:
def export_csv(self, path): import csv with open(path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) headers = [ self.table_result.horizontalHeaderItem(i).text() for i in range(self.table_result.columnCount()) ] writer.writerow(headers) for r in range(self.table_result.rowCount()): row = [ self.table_result.item(r, c).text() if self.table_result.item(r, c) else "" for c in range(self.table_result.columnCount()) ] writer.writerow(row)utf-8-sig编码会写入 BOM 头,Excel 双击打开时能正确识别 UTF-8,不会出现列挤在一列的问题。注意从表格导出时,每个单元格都要判断 item 是否为None,空单元格取.text()会直接报异常,这个小地方很容易翻车。
启动自检也值得加。程序启动时先select 1探一下连接,连不上就把错误信息弹到状态栏,而不是让用户打开后看到一个空窗口。我在 traceback 里抓到连接异常后,会在状态栏显示“连接失败:文件不存在或路径错误”,比默认报错信息友好得多。
我的习惯是:每接一个新库,先打开这个小工具,看一眼表结构和行数,再写查询;同事说数据不对,第一反应也是拿它跑一遍。做工具不是炫技,是把重复劳动压到最低。这套 PyQt5 的写法我改了三轮才算真正顺手,第一次写也栽在查询卡界面和锁库上。如果你照着这个结构做一版,大概率能绕开我踩过的坑。希望帮到你。
本文还有配套的精品资源,点击获取