全栈项目存储层重构:用SQLite替换JSON文件实战指南
2026/9/9 15:30:24 网站建设 项目流程

这次我们来看一个全栈项目里很实际的拐点:存储层重构。项目做到中期,数据处理早就不是几个 JSON 文件的简单读写了,查询、筛选、排序、多用户、批量导入,每个需求都在给存储层加压。这个阶段把存储层换掉,最平滑的方案就是 SQLite。

先给结论。SQLite 不需要单独安装数据库服务,不需要配置账号和端口,数据就放在单个文件里,Python 标准库自带的sqlite3模块可以直接操作。对一个中小型全栈项目来说,查询、事务、索引、批量写入这些能力完全够用,而且 SQL 语法和 MySQL、PostgreSQL 高度接近,后期真要换数据库,迁移成本也小。

这篇文章会按一条完整路线走:先准备环境和数据库文件,再设计存储层重构方案,接着封装数据访问层,把旧 JSON 数据迁进 SQLite,最后通过 Web API 把读写能力暴露出来,并补上功能测试、性能观察和常见问题排查。适合刚完成全栈入门、正在做自己的完整项目,或者想弄清楚“项目里到底该怎么正确引入数据库”的读者。

1. 核心能力速览

能力项说明
项目类型零到全栈系列中的重构实战
核心主题用 SQLite 替换原文件/内存存储层
数据能力SQL 查询、事务、索引、外键、批量写入
部署方式嵌入式数据库,无需独立服务进程
支持平台Windows / macOS / Linux
Web API可配合 Flask / FastAPI 等框架对外提供接口
批量任务支持executemany批量写入
可视化工具DB Browser for SQLite 等
适合场景全栈项目的本地持久化、中小型业务数据存储、教程项目

从重构的角度看,SQLite 在这里承担三个角色。第一,统一数据访问入口,业务层不再各自读写文件;第二,提供标准 SQL 能力,后续升级数据库时迁移成本可控;第三,把持久化变成强制行为,应用重启后数据不会丢。下面所有章节都会围绕这三个角色展开。

也许你会问:为什么不直接用 MySQL 或 PostgreSQL?答案是看阶段。开发期、原型期、个人项目,SQLite 的零运维特点能帮你省掉大量时间;真正出现多进程高并发写入瓶颈时,再迁移到服务型数据库,SQL 基础也能平移。这不是二选一,而是一条升级路径。

2. 适用场景与使用边界

SQLite 适合的场景很明确:个人项目、教程项目、公司内部工具、中小型 Web 应用的原型阶段,以及单机工具软件。如果你的项目是这几个类型,引入 SQLite 基本是零负担的——不需要运维,不需要单独部署,数据库文件跟着项目走,备份就是复制一个文件。

不适合的场景也要说清楚。高并发写入场景,比如多进程同时频繁写同一个数据库文件,SQLite 的串行写模型会成为瓶颈;需要细粒度行级权限控制的场景,SQLite 没有完整的权限体系;数据量到 TB 级或需要分布式部署的场景,SQLite 也不合适。这些情况应该直接选 PostgreSQL 或 MySQL。

再补充一条安全边界。无论用哪种数据库,涉及用户数据、业务数据和敏感资料时,都要先确认授权和合规要求。不要把包含真实用户数据的 db 文件提交到公开仓库,不要在不确定用途的前提下把敏感信息落到明文数据库里。本文所有代码和配置都建议在测试环境跑通后,再考虑应用到生产。

3. 环境准备与前置条件

3.1 本机 Python 与 sqlite3 检查

SQLite 最大的优势是零依赖。Python 3 自带sqlite3模块,属于官方标准库,不需要pip install任何第三方包。环境准备只需要两步:确认本机 Python 版本和sqlite3模块可用,以及规划一个清晰的目录结构。

先用一段代码确认环境:

import sys import sqlite3 print("Python 版本:", sys.version) print("SQLite 版本:", sqlite3.sqlite_version)

如果有输出版本号,说明可以直接进入下一节。如果这里报ModuleNotFoundError,说明当前 Python 环境不完整,建议重新安装 Python 3,不要继续往下走。

3.2 图形化管理工具(可选)

命令行不是必须的,但可视化工具能让你快速看到表结构和数据。常用的开源工具是 DB Browser for SQLite,跨平台支持,下载安装后打开数据库文件即可查看表、执行 SQL、导入导出数据。也可以直接使用 PyCharm、VS Code 里的 SQLite 插件。这个工具和 Python 代码互不冲突,主要用于调试和数据检查,不影响运行逻辑。

3.3 项目目录规划

重构前先规划好目录,避免数据库文件和旧数据、源码混在一起,后面找问题会非常痛苦。推荐的结构是这样:

project/ ├── app.py # Web 入口 ├── db.py # 数据库连接管理(本次新增) ├── todo_repo.py # 数据访问层(本次新增) ├── init_db.py # 建表脚本(本次新增) ├── data/ │ └── app.db # SQLite 数据库文件 ├── old_data/ │ └── data.json # 旧 JSON 数据(迁移后归档) └── requirements.txt

这是一个通用目录结构,实际项目按你自己的模块名调整。关键点是数据库文件、旧数据、源码分开管理,这样后续做备份和迁移都方便。

4. 安装部署与启动方式

4.1 初始化数据库与建表

SQLite 的“启动”不像 MySQL 那样要起服务。所谓的“启动”,其实就是建立连接并执行建表语句。以下代码会在data目录里生成app.db文件,并创建一个todos表:

# init_db.py import sqlite3 import os db_dir = "data" db_path = os.path.join(db_dir, "app.db") os.makedirs(db_dir, exist_ok=True) conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, completed INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now', 'localtime')) ) """) conn.commit() print("数据库创建成功:", db_path) conn.close()

这里用IF NOT EXISTS,保证脚本重复执行不会报错。AUTOINCREMENT让 id 自增,created_at自动取当前本地时间。实际项目里如果还有用户表、标签表、分类表,请按业务字段继续补充建表语句,示例的todos表只是用来演示一条完整的重构链路。

4.2 命令行验证

Python 创建完 db 文件后,可以用命令行工具直接查看。macOS 和 Linux 一般自带sqlite3,Windows 没有默认命令行,可以用图形工具,或者在 Python 里执行查询。命令行方式:

sqlite3 data/app.db ".tables" sqlite3 data/app.db ".schema todos"

如果本机没有sqlite3命令,直接用 Python 验证:

import sqlite3 conn = sqlite3.connect("data/app.db") rows = conn.execute( "SELECT name FROM sqlite_master WHERE type='table'" ).fetchall() print(rows) conn.close()

看到[('todos',)]这样的输出,说明建表成功。

5. 存储层重构方案

5.1 重构目标

假设项目原来的存储层是这样的:所有数据保存在一个data.json文件里;每个功能模块自己负责读写文件;查询、去重、排序都在 Python 里用列表循环完成;没有事务概念,写文件的时候可能互相覆盖丢失。这是很多全栈教程项目中期阶段的典型状态。

这次重构的目标有三条。第一,数据统一由数据库管理,业务模块通过数据访问层读写,不再直接碰文件;第二,提供标准 SQL 查询能力,把筛选、排序、统计交给数据库处理,而不是在 Python 里循环;第三,对外保持接口稳定,让上层调用方尽量不感知底层变化,减少业务代码改动。

5.2 重构步骤

整个重构过程可以拆成六步:

  1. 评估现状,找出所有读取和写入数据的代码位置,列一份清单。
  2. 设计表结构,把原来的对象字段映射成数据库字段,明确主键、外键、默认值。
  3. 编写建表脚本,集中管理CREATE TABLE,不散落在业务代码里。
  4. 实现数据访问层,提供create / read / update / delete函数。
  5. 迁移旧数据,把 JSON 批量导入数据库,并验证数量一致。
  6. 替换调用点,让业务模块走 repo 层函数,删除原来的文件读写逻辑。

这个过程的要点是“接口不变、底层替换”。上层调用方还是调用list_todos(),但函数内部的实现从读 JSON 变成了查数据库,改动影响面小,风险也可控。

5.3 数据访问层职责划分

引入数据库后最怕出现另一种乱:SQL 满天飞。路由层写 SQL,业务逻辑写 SQL,模板渲染前也要拼 SQL,这比文件存储还难维护。所以推荐把数据库操作集中到一个 repo 文件里,Web 层和业务层不直接写 SQL。

推荐的文件职责:db.py负责连接管理、事务和通用操作;todo_repo.py负责todos表的 CRUD,只暴露业务函数。这个分层在后续换 ORM(比如 SQLAlchemy)时,影响面也小。

6. 数据访问层封装

6.1 连接管理与事务

先写一个db.py,用上下文管理器统一管理连接。这样做的好处是:函数退出时自动 commit,出错自动 rollback,而且连接一定被关闭,不会出现句柄泄漏,也不会出现“数据看着写进去了,重启后丢失”的问题。

# db.py import sqlite3 from contextlib import contextmanager DB_PATH = "data/app.db" @contextmanager def get_conn(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close()

三个细节值得说明。row_factory = sqlite3.Row让查询结果可以按字段名访问,转 dict 更方便;PRAGMA foreign_keys = ON开启外键约束,SQLite 默认不启用,需要每次连接设置;上下文管理器在with块正常结束后 commit,异常后 rollback,确保事务边界清晰。

6.2 CRUD 函数实现

接着写todo_repo.py,把插入、查询、更新、删除都封装成函数。注意所有 SQL 都使用?占位符传参,而不是字符串拼接,这是防止 SQL 注入的基本要求。业务层输入的内容即使包含引号、分号,也不会被当成 SQL 执行。

# todo_repo.py from db import get_conn def create_todo(title: str) -> int: """新增一条待办,返回自增 id""" with get_conn() as conn: cursor = conn.execute( "INSERT INTO todos (title) VALUES (?)", (title,) ) return cursor.lastrowid def list_todos(): """查询所有待办,按 id 倒序""" with get_conn() as conn: rows = conn.execute( """ SELECT id, title, completed, created_at FROM todos ORDER BY id DESC """ ).fetchall() return [dict(row) for row in rows] def update_todo(todo_id: int, title: str = None, completed: bool = None): """按 id 更新待办,title 和 completed 可传一个或两个""" with get_conn() as conn: if title is not None: conn.execute( "UPDATE todos SET title = ? WHERE id = ?", (title, todo_id) ) if completed is not None: conn.execute( "UPDATE todos SET completed = ? WHERE id = ?", (1 if completed else 0, todo_id) ) def delete_todo(todo_id: int): """按 id 删除待办""" with get_conn() as conn: conn.execute("DELETE FROM todos WHERE id = ?", (todo_id,)) def count_todos() -> int: """统计待办总数""" with get_conn() as conn: row = conn.execute( "SELECT COUNT(*) AS total FROM todos" ).fetchone() return row["total"]

update_todo的设计很实用:两个参数都可选,前端传哪个就更新哪个,避免一次更新把所有字段全查出来再写回去。count_todos在迁移验证时会用到。

7. 旧数据迁移与批量导入

7.1 JSON 数据读取

重构过程中,旧数据不能丢。假设旧数据在old_data/data.json,结构大概是这样的数组:

[ {"title": "学习 SQLite", "completed": 0}, {"title": "重构存储层", "completed": 1} ]

迁移脚本要做的三件事:读取 JSON、写入数据库、验证数量。这一步同时也是全栈项目里最常见的批量导入任务,逻辑不复杂,但很容易在编码和重复执行上出错。

7.2 批量插入

批量插入尽量用executemany,不要用for循环逐条executeexecutemany会把多条 INSERT 一次性绑定,写入速度比逐个提交快很多,尤其数据量到几千条以上时,差异非常明显。

# migrate.py import json import sqlite3 JSON_PATH = "old_data/data.json" DB_PATH = "data/app.db" def migrate_json_to_sqlite(json_path: str = JSON_PATH, db_path: str = DB_PATH): with open(json_path, "r", encoding="utf-8") as f: old_data = json.load(f) conn = sqlite3.connect(db_path) cursor = conn.cursor() # 建表(与 init_db.py 保持一致) cursor.execute(""" CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, completed INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now', 'localtime')) ) """) rows = [ (item["title"], int(item.get("completed", 0))) for item in old_data ] cursor.executemany( "INSERT INTO todos (title, completed) VALUES (?, ?)", rows ) conn.commit() count = cursor.execute( "SELECT COUNT(*) FROM todos" ).fetchone()[0] print(f"迁移完成,todos 表当前共 {count} 条数据") conn.close() if __name__ == "__main__": migrate_json_to_sqlite()

如果源 JSON 字段名和示例不一致,需要把映射逻辑换成自己项目的字段名。执行迁移前,先确认数据库里没有重复数据;如果可能重复执行,可以使用INSERT OR IGNORE并配合唯一索引,具体去重策略根据业务决定。

7.3 迁移验证与验证结果

迁移后不要急着删旧文件,先验证。通过count_todos()查总数和旧数据条数是否一致;再随机抽查几条记录,对比 title 和 completed 字段;然后让应用以新存储层启动,确认页面能正常显示数据;最后把旧 JSON 文件改名归档,保留一段时间再删除。

from todo_repo import count_todos, list_todos print("总数:", count_todos()) for item in list_todos()[:5]: print(item["id"], item["title"], item["completed"])

只有确认这些结果都正常,才可以把原来的文件读写逻辑删掉。保留旧文件不是胆小,而是给回滚留一条退路。

8. Web API 衔接示例

8.1 Flask 接口改造

如果全栈项目有 Web 后端,接下来就是把原来的路由处理函数从“直接读写文件”改成“调用 repo 层函数”。这里以 Flask 为例,FastAPI 的用法类似,把装饰器和请求对象换成 FastAPI 风格即可。核心变化是路由函数不再关心数据存在哪里,只调用todo_repo里的函数。

# app.py from flask import Flask, request, jsonify from todo_repo import create_todo, list_todos from todo_repo import update_todo, delete_todo app = Flask(__name__) @app.route("/api/todos", methods=["GET"]) def get_todos(): return jsonify(list_todos()) @app.route("/api/todos", methods=["POST"]) def add_todo(): data = request.get_json(force=True) title = (data.get("title") or "").strip() if not title: return jsonify({"error": "title 不能为空"}), 400 todo_id = create_todo(title) return jsonify({"id": todo_id, "title": title}), 201 @app.route("/api/todos/<int:todo_id>", methods=["PUT"]) def edit_todo(todo_id): data = request.get_json(force=True) update_todo( todo_id, title=data.get("title"), completed=data.get("completed") ) return jsonify({"ok": True}) @app.route("/api/todos/<int:todo_id>", methods=["DELETE"]) def remove_todo(todo_id): delete_todo(todo_id) return jsonify({"ok": True}) if __name__ == "__main__": app.run(host="127.0.0.1", port=5000, debug=True)

这段代码里可以看到,路由层只做 HTTP 参数解析和返回值格式化,所有数据操作用一行函数调用完成。后续即使把 SQLite 换成 PostgreSQL,也只需要替换todo_repo的实现,路由层几乎不用动。

8.2 curl 与 Python 请求验证

启动 Flask 服务后,可以用 curl 直接测接口。先启动服务:

python app.py

然后打开另一个终端执行:

# 查询列表 curl http://127.0.0.1:5000/api/todos # 新增一条 curl -X POST http://127.0.0.1:5000/api/todos \ -H "Content-Type: application/json" \ -d '{"title": "测试 SQLite 接口"}' # 更新 curl -X PUT http://127.0.0.1:5000/api/todos/1 \ -H "Content-Type: application/json" \ -d '{"completed": true}' # 删除 curl -X DELETE http://127.0.0.1:5000/api/todos/1

也可以用 Python requests 做一次完整链路测试:

import requests base = "http://127.0.0.1:5000/api/todos" # 新增 r = requests.post(base, json={"title": "用 requests 测试"}) print("POST", r.status_code, r.json()) # 查询 r = requests.get(base) print("GET", r.status_code, len(r.json())) # 更新 todo_id = r.json()[0]["id"] r = requests.put(f"{base}/{todo_id}", json={"completed": True}) print("PUT", r.status_code, r.json()) # 删除 r = requests.delete(f"{base}/{todo_id}") print("DELETE", r.status_code, r.json())

接口能跑通,后面就可以把前端页面指向这些 API,或者把批量数据导入、定时任务都接到这个 repo 层上。

9. 功能测试与效果验证

9.1 测试用例清单

重构之后,必须验证的不只是“接口能通”,而是“数据链路全流程正确”。建议至少覆盖下面这些用例:

测试项操作预期结果
建表执行init_db.pydata/app.db生成,todos表存在
新增调用create_todo返回自增 id,数据库中多一条记录
查询调用list_todos返回列表,字段完整
更新修改 title / completed对应记录字段发生变化
删除删除一条记录记录消失,其他记录不受影响
持久化重启应用后再查询重启前写入的数据仍然存在
事务回滚故意在事务中抛异常已执行的 INSERT 被回滚
批量导入导入旧 JSON数据条数一致,字段无误

如果项目里有多张表,还需要额外验证联表查询。SQLite 支持INNER JOINLEFT JOIN,一旦涉及联表,要确认关联字段的索引以及外键约束是否生效。可以用EXPLAIN QUERY PLAN查看执行计划,确认是否走了索引,避免全表扫描拖慢接口。

sqlite3 data/app.db \ "EXPLAIN QUERY PLAN SELECT * FROM todos t LEFT JOIN tags tg ON t.tag_id = tg.id;"

这段 SQL 只是演示语句,实际字段以你的表结构为准。

9.2 持久化与事务验证

持久化验证很关键,因为这是从文件存储换成数据库最核心的收益之一。操作方法是:调用create_todo写入一条数据,重启 Web 服务或重新执行脚本,再查询数据是否还在。

from todo_repo import create_todo, list_todos create_todo("重启后这条数据应该还在") for item in list_todos(): print(item["id"], item["title"], item["completed"])

事务回滚验证:

import sqlite3 conn = sqlite3.connect("data/app.db") try: conn.execute("INSERT INTO todos (title) VALUES ('事务测试')") raise RuntimeError("手动抛错") except RuntimeError: conn.rollback() print("已回滚") count = conn.execute( "SELECT COUNT(*) FROM todos WHERE title='事务测试'" ).fetchone()[0] print("事务测试数据存在条数:", count) conn.close()

如果事务回滚生效,最后的 count 应该是 0。生产环境不要用这种裸连接写法,这里只是为了把事务语义演示清楚。

9.3 判断重构是否成功

一个很实用的判断标准:把旧存储层代码删掉或注释掉,整个项目还能正常工作。如果你的业务代码仍然在直接读data.json,说明重构还没改干净。可以全局搜索openjson.loadjson.dump这些关键字,逐一确认是否都替换成 repo 层调用。

另一个判断标准是数据一致性。在测试环境里模拟一次崩溃或异常退出,再重启应用,看看已提交的数据是否完整、未提交的脏数据是否被回滚。这个测试通过,重构的核心目标就达成了。

10. 资源占用与性能观察

10.1 数据库文件大小

SQLite 数据都保存在单个 db 文件里,文件大小能直观反映数据量。不同平台查看方式不同,用 Python 打印最通用:

import os size = os.path.getsize("data/app.db") print(f"app.db 大小: {size} bytes = {size / 1024:.1f} KB")

实际大小取决于数据量和是否执行过VACUUM。删除大量数据后文件不会自动缩小,可以定期执行VACUUM;回收空间。

10.2 并发与 WAL 模式

SQLite 默认的 journal 模式在并发读写时可能出现database is locked。如果你的 Web 服务有多个请求同时写库,建议开启 WAL 模式:

conn.execute("PRAGMA journal_mode = WAL")

WAL 模式下,读操作和写操作可以并发执行,对中小型 Web 应用非常友好。这个设置不是只执行一次,而是每次连接都要确认,可以放到db.pyget_conn里。

10.3 批量写入性能对比

批量写入是常见的性能瓶颈。一个简单的观察方法是插入 1000 条数据,分别用单条循环和executemany测试:

import sqlite3 import time conn = sqlite3.connect("data/app.db") cursor = conn.cursor() # 方案一:逐条插入,包在同一个事务里 start = time.time() conn.execute("BEGIN") for i in range(1000): cursor.execute( "INSERT INTO todos (title) VALUES (?)", (f"任务{i}",) ) conn.commit() print("逐条插入 1000 条耗时:", time.time() - start) # 方案二:executemany 批量插入 start = time.time() cursor.executemany( "INSERT INTO todos (title) VALUES (?)", [(f"批量任务{i}",) for i in range(1000)] ) conn.commit() print("executemany 插入 1000 条耗时:", time.time() - start) conn.close()

注意上面代码里先BEGIN再逐条插入,是为了把多条 INSERT 包在一个事务里;如果每执行一条就 commit 一次,性能会差得更多。具体数字会因机器和磁盘不同而不同,建议你自己跑一次,重点不是记一个绝对数字,而是掌握这个观察方法,后续在真实项目里做批量任务时知道怎么评估性能。

10.4 内存与 CPU 观察

全栈项目运行期的内存和 CPU 数据,和你的 Web 框架、业务代码关系更大。SQLite 本身是文件 IO 密集型,不是常驻进程,所以它不像 MySQL 那样独占内存。要观察项目整体资源占用,可以用系统自带工具:Windows 的任务管理器,macOS 的活动监视器,Linux 的topps。重点看服务进程有没有异常增长、有没有大量连接积压。

11. 常见问题与排查方法

这一节把重构成 SQLite 后最常遇到的问题整理成表,适合收藏备用。

问题现象可能原因排查方式解决方案
查询报 no such table建表脚本未执行,或表名/字段名不一致打印sqlite_master查看表列表先执行建表脚本,确认表名大小写
报 database is locked多进程或多线程同时写一个 db 文件检查是否有进程持有连接开启 WAL,设置timeout=5,避免长事务
插入数据后查询不到没有 commit检查代码中是否调用conn.commit()使用上下文管理器统一提交/回滚
中文数据乱码读写文件编码不一致检查 JSON 文件编码和终端编码读写时指定encoding="utf-8"
db 文件被占用无法删除有连接未关闭检查进程是否存活关闭所有连接,结束残留进程
外键约束不生效SQLite 默认关闭外键打印PRAGMA foreign_keys每次连接执行PRAGMA foreign_keys = ON
迁移导入数据重复重复执行迁移脚本查询表中总条数导入前清空表,或加唯一索引配合INSERT OR IGNORE
文件路径找不到工作目录不对打印os.getcwd()和 db 路径使用绝对路径,或统一从项目根目录启动

针对最常遇到的两个问题再展开一下。

database is locked。SQLite 对写入是串行的,多个连接同时写同一个文件会锁库。短期的解决方案是设置sqlite3.connect(db_path, timeout=5)让写操作等待锁;长期方案是切换 WAL 模式,并把连接放在短事务里,不要在事务中做耗时的网络请求。如果你的服务是多进程模型,还要确认每个进程用的是同一个 db 文件路径,避免出现多个拷贝文件互相看不见数据的问题。

no such table。很多人第一次跑会漏掉初始化脚本。建议把建表逻辑独立成一个init_db.py,部署和测试时先执行它。否则业务代码访问时,数据库文件是新的空文件,里面根本没有表。这个错误还有一个变种:表名和字段名大小写不一致。SQLite 的表名和字段名在某些情况下区分大小写,最稳妥的做法是建表时统一小写,代码里也一律用小写。

12. 最佳实践与使用建议

12.1 项目工程化建议

把这几个习惯固定下来,项目会少很多隐蔽的 bug。

第一,所有 SQL 都走参数化查询。业务层输入的字符串永远不要用f"SELECT ... WHERE name='{value}'"这种写法去拼接,这是最基本的防注入手段,也是全栈项目的合规底线。

第二,连接统一管理。写一个get_conn上下文管理器,所有函数都用with get_conn() as conn包裹,commit、rollback、close 全部自动处理。不要在多个函数里各写一遍sqlite3.connect,那样会漏掉关闭,导致文件被占用。

第三,建表语句集中管理。不要散落在业务代码里,数据库 schema 变化时应该走显式的迁移脚本,而不是在某个请求里顺手改表。

第四,重要表加索引。比如按 completed 筛选、按 created_at 排序的场景,建索引后查询差距明显:

CREATE INDEX idx_todos_completed ON todos(completed); CREATE INDEX idx_todos_created_at ON todos(created_at);

12.2 数据与合规建议

使用 SQLite 存储真实业务数据时,要养成三个习惯。

定期备份。SQLite 备份最简单的方式是复制 db 文件,也可以使用sqlite3的在线备份 API;不要把包含真实用户数据的数据库文件提交到公开仓库,开发环境用一个app-dev.db,生产环境用另一个不受版本控制的 db 文件;涉及敏感数据时,不管用哪种数据库,都必须先确认授权范围,做好访问控制和定时清理。

12.3 性能与维护习惯

数据量增长后,建议定期执行:

VACUUM; ANALYZE;

VACUUM可以回收文件空间,ANALYZE更新查询优化器使用的统计信息。这两个命令在数据量大、频繁增删的场景下更值得重视。另外,SQLite 的备份、导出、导入都可以用sqlite3命令完成,遇到数据问题时多一个处理工具总是好的。

13. 总结与下一步

这次重构的核心思路可以归纳成一句话:业务层尽量不碰 SQL 和文件,所有数据操作都收口到数据访问层。你最先应该验证的是三个点:建表脚本能不能稳定执行、CRUD 函数能不能跑通、重启进程后数据还在不在。这三个点确认了,重构就等于完成了一大半。

最容易踩的坑也有三个:一是建表脚本没有执行就查表,报no such table;二是连接没有 commit,数据看着写进去了但重启消失;三是多进程写库没有开启 WAL,偶发database is locked

下一步可以继续做三件事:把 repo 层换成 SQLAlchemy,提前把 ORM 接入路径铺好;引入迁移工具 Alembic,让表结构变更可以回滚和追溯;在集成测试里把所有 CRUD 和事务用例自动化,防止后续改版回退。

SQLite 不是一个“临时凑合”的方案。很多轻量级全栈项目、桌面工具、嵌入式应用,生产环境也长期跑在 SQLite 上。这次重构把存储层理顺之后,后续不管是继续加功能、加接口,还是换更重的数据库,路径都会比现在清楚得多。建议把上面的初始化脚本、repo 层代码和测试用例保存成模板,下一个项目直接复用。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询