1. 为什么要在重构时考虑 SQLite1.1 项目重构的本质很多开发者把“重构”理解为“重写”但实际上两者差别很大。重写是把代码推倒重来风险高、周期长重构是在不改变外部功能的前提下调整内部结构让代码更容易阅读、更容易扩展、也更容易维护。在“零到全栈”系列的前几个阶段项目通常以“能跑通”为第一目标。数据可能写在 JSON 文件、内存变量或者列表里。等到业务功能越来越多你会发现几个比较明显的信号每次启动都要把整个数据文件加载到内存数据量一大就卡。多个功能都要读写同一份数据代码里到处是open()、json.load()、json.dump()。想查询某一条记录只能遍历整个列表时间复杂度是 O(n)。并发写入时容易出现数据覆盖甚至文件损坏。这些信号出现时就是引入数据库的合适时机。但“零到全栈”项目的规模通常还没到需要 MySQL、PostgreSQL 这种独立数据库服务的程度。这时候 SQLite 是非常合适的选择。1.2 SQLite 是什么SQLite 是一个嵌入式关系型数据库它不是一个独立的服务器进程而是以库文件的形式直接嵌入到应用程序中。用一句话概括SQLite 就是“一个文件 一个库”你的所有表、索引、数据都存放在这个文件里。SQLite 的特点非常适合“零到全栈”阶段的开发者零配置不需要单独安装数据库服务。轻量级整个数据库引擎只有几百 KB。支持标准 SQL绝大部分 SQL 语法在 SQLite 中可以直接使用。跨平台Windows、Linux、macOS 都能运行。几乎所有主流语言都有对应的驱动Python 更是直接内置了sqlite3模块。你可能会问那它有缺点吗有。SQLite 不适合高并发写入场景也不适合多机分布式访问。但在单机应用、桌面工具、移动端 App、中小型 Web 项目的初级阶段SQLite 的性能完全够用。1.3 什么场景适合在重构时引入 SQLite结合“重构”这个主题以下场景非常适合把原有文件存储改造为 SQLite数据量超过几百条并且需要频繁查询、排序、分组。多个模块需要共享同一份数据但又不想手动维护文件锁。需要保证数据一致性例如一次写入多条记录要么全部成功要么全部失败。需要做数据版本迁移例如从 v1 的数据结构升级到 v2。数据需要导出、备份、恢复SQLite 的单文件特性让备份变得非常方便。在机房管理系统、工具箱类工具软件中SQLite 非常常见。比如机房上机记录、用户操作日志、软件配置信息、离线缓存都是 SQLite 的典型应用场景。理解 SQLite 之后你再看这类项目的源码就不会感到陌生。2. 环境准备与版本说明2.1 操作系统与运行环境本文示例以 Windows 11 Python 3.10 为演示环境但所有代码都兼容 Linux 和 macOS。如果你的系统不同只需要注意文件路径的写法即可。Python 2.7 已经停止维护建议使用 Python 3.6 以上版本。本文代码基于 Python 3 语法编写。python --version如果你还没有安装 Python可以去 Python 官网下载安装包安装时记得勾选“Add Python to PATH”。2.2 SQLite 与 DB BrowserSQLite 本身是一个 C 语言库Python 自带的sqlite3模块已经封装了它所以你不需要额外安装 SQLite。但日常开发中我建议安装一个可视化工具DB Browser for SQLite。它是一个免费开源工具可以直观地查看表结构、浏览数据、执行 SQL 语句非常适合学习和调试。下载地址可以直接搜索“DB Browser for SQLite”官网选择对应系统的安装包即可。安装完成后用它可以打开.db文件。另外Navicat for SQLite 也是一款常见的图形化工具功能更丰富但它属于商业软件需要购买授权。不建议使用破解版本学习阶段使用 DB Browser for SQLite 完全足够。2.3 示例项目结构为了让重构案例更清晰我先给出项目的目标结构sqlite_refactor_demo/ ├── data/ │ ├── users_old.json # 原始 JSON 数据重构前 │ └── users.db # SQLite 数据库重构后 ├── src/ │ ├── db.py # 数据库访问层 │ ├── models.py # 数据模型 │ └── user_service.py # 业务层 └── main.py # 程序入口在重构过程中我们会保留原始 JSON 数据作为对比同时实现一个数据库访问层让业务代码不再直接操作文件。3. 重构前的项目状态分析3.1 原有实现JSON 文件存储假设我们之前实现了一个简单的用户管理系统所有用户数据保存在users_old.json中[ {id: 1, name: Alice, age: 25, city: Beijing}, {id: 2, name: Bob, age: 30, city: Shanghai}, {id: 3, name: Cathy, age: 28, city: Guangzhou} ]原来的业务代码大致如下import json def load_users(): with open(data/users_old.json, r, encodingutf-8) as f: return json.load(f) def save_users(users): with open(data/users_old.json, w, encodingutf-8) as f: json.dump(users, f, ensure_asciiFalse, indent4) def find_user_by_name(name): users load_users() for user in users: if user[name] name: return user return None def add_user(user): users load_users() users.append(user) save_users(users)3.2 现有问题这段代码在项目初期没什么问题但随着功能扩展问题逐渐显现每次查找都要读取整个文件。如果用户有 10 万条一次查找就要加载 10 万条数据到内存。写入是“整体覆盖”。每次增加一个用户都要把整个数组重新写入文件。写入过程中如果程序崩溃文件可能损坏。缺少数据约束。JSON 文件不能保证每条数据都有id、name等字段也不能保证id是唯一的。并发写入不安全。两个请求同时调用add_user后写入的会把先写入的覆盖掉。这些问题在真实业务中都非常致命。接下来我们通过重构把存储层从 JSON 文件替换为 SQLite。4. SQLite 核心知识拆解4.1 创建表在 SQLite 中创建表使用CREATE TABLE语句CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER NOT NULL, city TEXT NOT NULL, created_at TEXT DEFAULT (datetime(now)) );关键参数解释INTEGER PRIMARY KEY AUTOINCREMENT自增主键保证每条记录有唯一 ID。TEXT NOT NULL字段类型为文本且不允许为空。DEFAULT (datetime(now))如果插入时没有指定该字段会自动填充当前时间。IF NOT EXISTS如果表已存在则不重复创建避免重复执行脚本时出错。需要注意SQLite 的数据类型是“弱类型”它会根据值自动判断类型。但建议在结构设计阶段就明确类型方便后续维护。4.2 增删改查插入数据INSERT INTO users (name, age, city) VALUES (Alice, 25, Beijing);查询数据SELECT id, name, age, city FROM users WHERE city Beijing;更新数据UPDATE users SET age 26 WHERE name Alice;删除数据DELETE FROM users WHERE id 1;这里要特别强调UPDATE 和 DELETE 语句的WHERE子句必须谨慎。如果漏写WHERE会更新或删除表中所有数据。真实项目中上线前一定要确认 SQL 语句的范围。4.3 事务与参数绑定在 Python 的sqlite3模块中执行写入操作时要注意两件事事务和参数绑定。SQLite 默认是自动提交模式但为了确保数据一致性更好的做法是显式控制事务import sqlite3 def add_user(name, age, city): conn sqlite3.connect(data/users.db) try: cursor conn.cursor() cursor.execute( INSERT INTO users (name, age, city) VALUES (?, ?, ?), (name, age, city) ) conn.commit() except Exception as e: conn.rollback() raise e finally: conn.close()这里使用了?占位符。千万不要用字符串拼接 SQL否则会引发 SQL 注入风险。?占位符会把参数和 SQL 语句分离由数据库驱动负责安全的参数转义。事务的核心逻辑是先执行 SQL如果中间出现异常调用rollback()回滚确保数据不会处于“插入了一半”的状态。4.4 索引当数据量变大后查询性能会明显下降。这时候可以给常用查询字段加索引CREATE INDEX idx_users_city ON users(city);加了索引后WHERE city Beijing这类查询会显著加快。但索引不是越多越好因为每次插入和更新数据都要额外维护索引。一般在确定是高频查询字段的情况下才加索引。5. 完整重构实战5.1 设计表结构在开始写代码前先设计表结构。我们继续以用户管理为例设计users表字段名类型约束说明idINTEGERPRIMARY KEY AUTOINCREMENT用户 IDnameTEXTNOT NULL用户名ageINTEGERNOT NULL年龄cityTEXTNOT NULL所在城市created_atTEXTDEFAULT datetime(now)创建时间同时创建一个初始化数据库的脚本# 文件路径src/db.py import sqlite3 import os DB_PATH os.path.join(os.path.dirname(os.path.dirname(__file__)), data, users.db) def get_connection(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_connection() try: cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER NOT NULL, city TEXT NOT NULL, created_at TEXT DEFAULT (datetime(now)) ) ) conn.commit() finally: conn.close()conn.row_factory sqlite3.Row可以让查询结果的每一行像字典一样按字段名访问同时支持索引访问代码可读性更好。5.2 编写数据库访问层数据库访问层负责所有 SQL 操作业务层不需要关心 SQL 细节。# 文件路径src/user_repository.py from src.db import get_connection def get_all_users(): conn get_connection() try: cursor conn.cursor() cursor.execute(SELECT id, name, age, city, created_at FROM users ORDER BY id) rows cursor.fetchall() return [dict(row) for row in rows] finally: conn.close() def find_user_by_name(name): conn get_connection() try: cursor conn.cursor() cursor.execute( SELECT id, name, age, city, created_at FROM users WHERE name ?, (name,) ) row cursor.fetchone() return dict(row) if row else None finally: conn.close() def add_user(user): conn get_connection() try: cursor conn.cursor() cursor.execute( INSERT INTO users (name, age, city) VALUES (?, ?, ?), (user[name], user[age], user[city]) ) conn.commit() return cursor.lastrowid except Exception: conn.rollback() raise finally: conn.close() def update_user_age(user_id, new_age): conn get_connection() try: cursor conn.cursor() cursor.execute( UPDATE users SET age ? WHERE id ?, (new_age, user_id) ) conn.commit() return cursor.rowcount 0 except Exception: conn.rollback() raise finally: conn.close() def delete_user(user_id): conn get_connection() try: cursor conn.cursor() cursor.execute(DELETE FROM users WHERE id ?, (user_id,)) conn.commit() return cursor.rowcount 0 finally: conn.close()这里每个函数的模式非常统一建立连接、执行 SQL、提交事务、关闭连接。把重复的流程封装好后业务层会干净很多。5.3 业务层改造原来的业务代码直接操作 JSON 文件现在改成调用user_repository中的函数。# 文件路径src/user_service.py from src import user_repository def list_users(): return user_repository.get_all_users() def search_user(name): return user_repository.find_user_by_name(name) def create_user(name, age, city): user { name: name, age: age, city: city } new_id user_repository.add_user(user) return {id: new_id, **user} def modify_user_age(user_id, new_age): success user_repository.update_user_age(user_id, new_age) if success: return {message: 更新成功} return {message: 用户不存在}, 404 def remove_user(user_id): success user_repository.delete_user(user_id) if success: return {message: 删除成功} return {message: 用户不存在}, 404对比原来的代码业务层不再关心数据是怎么存储的。将来如果要从单机版升级为 MySQL只需要替换user_repository的实现业务层基本不用改动。这就是分层重构的价值通过“数据访问层”隔离了存储细节。5.4 主程序入口与数据迁移为了让旧数据也能继续使用我们需要写一个“导入旧 JSON 数据”的函数# 文件路径src/migrate.py import json from src import user_repository def import_users_from_json(json_path): with open(json_path, r, encodingutf-8) as f: users json.load(f) for user in users: exists user_repository.find_user_by_name(user[name]) if exists is None: user_repository.add_user(user) else: print(f跳过重复用户: {user[name]})主程序入口# 文件路径main.py from src.db import init_db from src.migrate import import_users_from_json from src import user_service def main(): # 1. 初始化数据库 init_db() # 2. 导入旧 JSON 数据 import_users_from_json(data/users_old.json) # 3. 查询全部用户 users user_service.list_users() for user in users: print(user) # 4. 新增一个用户 new_user user_service.create_user(David, 35, Shenzhen) print(新增用户:, new_user) # 5. 搜索用户 found user_service.search_user(Alice) print(搜索结果:, found) # 6. 更新年龄 result user_service.modify_user_age(found[id], 26) print(更新结果:, result) if __name__ __main__: main()运行方式cd sqlite_refactor_demo python main.py预期输出结果类似跳过重复用户: Alice 跳过重复用户: Bob 跳过重复用户: Cathy {id: 1, name: Alice, age: 25, city: Beijing, created_at: 2025-01-01 10:00:00} {id: 2, name: Bob, age: 30, city: Shanghai, created_at: 2025-01-01 10:00:01} {id: 3, name: Cathy, age: 28, city: Guangzhou, created_at: 2025-01-01 10:00:02} 新增用户: {id: 4, name: David, age: 35, city: Shenzhen} 搜索结果: {id: 1, name: Alice, age: 25, city: Beijing, created_at: 2025-01-01 10:00:00} 更新结果: {message: 更新成功}到这里原来的 JSON 文件存储已经被成功重构为 SQLite 存储。整个重构过程没有改变外部功能但内部的数据访问方式完全变了。6. 常见问题与排查思路6.1 database is locked错误现象执行写入操作时程序抛出sqlite3.OperationalError: database is locked。常见原因另一个连接已经持有了数据库的写锁尚未提交或关闭。多个连接同时写入SQLite 默认的锁等待时间太短。连接没有正确关闭导致文件被长期占用。解决思路确保每次操作完成后都关闭连接推荐使用with语句或finally。设置合理的超时时间sqlite3.connect(DB_PATH, timeout10)。对于需要频繁写入的场景考虑使用 WAL 模式在下一节会讲解。如何避免统一封装数据库连接不要在每个函数里随意创建连接。6.2 插入中文数据后乱码错误现象数据库中中文显示为乱码比如???。常见原因在 Python 2 时代这个问题比较常见。Python 3 的字符串默认是 Unicodesqlite3模块默认支持 UTF-8一般不会出现乱码。如果使用DB Browser for SQLite打开数据库看到乱码需要检查是否选择了正确的编码方式。解决思路Python 3 项目中把源代码文件编码固定为 UTF-8。使用文件读写时显式指定encodingutf-8。查看数据时使用支持 UTF-8 的数据库工具。6.3 列表查询返回空结果错误现象明明数据库里有数据但SELECT查询返回空。常见原因查询条件写错比如明明写的是city Beijing表中存的是BEIJING。数据库连接指向了错误的文件路径。项目中使用相对路径时当前工作目录不同会导致找不到同一个数据库文件。忘记提交事务数据还在缓存中另一个连接看不到。解决思路使用DB Browser for SQLite打开数据库先确认表里是否有数据。打印出当前DB_PATH的绝对路径确认连接的是同一个文件。写入后调用conn.commit()。6.4 主键自增不连续错误现象删除某些行后再插入数据id不会复用被删除的编号而是继续增加。说明这是 SQLite 的正常行为。AUTOINCREMENT使用sqlite_sequence表来记录当前最大值即使删除了所有记录自增计数器也不会重置。如果业务上不要求 id 绝对连续不需要处理如果希望复用可以在插入前手动指定 id。7. 最佳实践与工程建议7.1 数据库文件备份SQLite 的备份非常简单因为整个数据库就是一个.db文件。推荐两种备份方式文件复制把.db文件复制一份放到备份目录中。但要注意一定要在程序不写数据库时复制否则可能复制到中间状态。在线备份 API使用 Python 的sqlite3.Connection.backup()方法可以在数据库运行期间做安全备份。import sqlite3 def backup_db(source_db, target_db): src sqlite3.connect(source_db) dst sqlite3.connect(target_db) try: src.backup(dst) finally: dst.close() src.close()7.2 事务粒度要小事务是保证数据一致性的重要机制但不要在一个事务中执行大量无关操作。事务时间越长锁持有时间越长其他连接等待的时间就越长。建议只把需要保证原子性的 SQL 放在同一个事务中。避免在事务中调用外部接口、等待用户输入、执行耗时计算。写完立即提交。7.3 安全边界问题虽然 SQLite 是单机文件数据库但同样存在安全问题禁止拼接 SQL 字符串。所有用户输入必须通过?占位符传入。数据库文件不要放在 Web 根目录下。否则一旦 Web 服务配置不当用户可以直接下载.db文件。敏感字段加密存储。如果表中包含密码、Token 等信息不要明文存储至少使用哈希算法处理。7.4 开启 WAL 模式WALWrite-Ahead Logging预写日志模式可以显著改善 SQLite 的并发读取性能。在 WAL 模式下读操作和写操作可以并发执行不会因为写锁阻塞读操作。conn sqlite3.connect(data/users.db) cursor conn.cursor() cursor.execute(PRAGMA journal_modeWAL;) cursor.execute(PRAGMA synchronousNORMAL;)WAL 模式会在数据库文件旁边生成users.db-wal和users.db-shm文件。备份时需要注意只复制.db文件可能丢失 WAL 中尚未合并的数据。建议使用备份 API或者在备份前先执行PRAGMA wal_checkpoint(TRUNCATE);把日志合并到主文件。7.5 数据库版本迁移项目迭代过程中表结构往往需要调整。建议在项目里维护一个SCHEMA_VERSION配置配合迁移脚本按版本号递增执行# 简单迁移思路 MIGRATIONS [ # 版本 1初始建表 CREATE TABLE users (...) , # 版本 2新增字段 ALTER TABLE users ADD COLUMN email TEXT , ]每次启动时检查当前数据库版本低于最新版本的依次执行迁移脚本最后更新版本记录。这样团队协作时每个人拉取代码后都能自动升级数据库结构。7.6 连接管理在 Web 应用中频繁创建和关闭连接有一定开销。可以维护一个全局连接对象但要注意线程安全。Python 的sqlite3模块默认情况下同一个连接在不同线程中使用需要设置check_same_threadFalse但不推荐随意使用。对于入门项目建议在数据访问层保持“一次操作建一次连接”的方式简单清晰不容易出现连接泄漏。等性能确实成为瓶颈后再考虑连接池方案。8. 总结与下一步通过这次重构我们把一个基于 JSON 文件存储的用户管理模块成功改造成基于 SQLite 的存储方案。这个过程中你至少应该掌握以下知识点SQLite 适合单机应用、中小型 Web 项目和工具类软件的持久化存储。使用sqlite3模块连接数据库、创建表、执行增删改查。使用?占位符防止 SQL 注入。事务控制的三件套commit()、rollback()、finally close()。分层重构的思路用数据访问层隔离存储实现业务层不要关心底层存储细节。数据库文件的备份、迁移和并发模式的工程化处理。下一步你可以继续探索用 Flask 或 FastAPI 把用户服务封装成 REST API真正实现“前后端分离”。把 SQLite 替换为 MySQL理解数据库驱动层的变化。深入学习 SQL 查询优化比如索引、EXPLAIN QUERY PLAN。学习数据库迁移工具例如 Alembic实现更规范的表结构变更管理。实际项目中建议从项目一开始就评估数据存储方案不要等项目跑起来再补数据库。但如果你的项目已经出现了内存列表和 JSON 文件越来越难维护的情况SQLite 是一个非常平滑的过渡方案。只要记住变更之前先备份写完 SQL 先确认WHERE提交事务前看清楚逻辑就不会出大问题。