在 Python 开发中，数据存储是绕不开的核心环节

在 Python 开发中数据存储是绕不开的核心环节。对于小型应用、原型开发、自动化脚本或测试环境来说引入 MySQL、PostgreSQL 这类客户端-服务端架构的数据库往往显得杀鸡用牛刀——需要安装服务、配置账号、管理端口部署成本高。而 SQLite 正是为解决这类轻量级数据持久化需求而生的。SQLite 是一款开源的嵌入式关系型数据库由 C 语言编写自 Python 2.5 起便作为标准库sqlite3模块内置于 Python 中无需任何额外安装即可直接使用。 整个数据库以单个.db文件形式存储在磁盘上支持完整的 SQL 语法、ACID 事务、索引、视图、触发器等功能。 本报告将从 SQLite 的核心特性出发系统讲解 Python 操作 SQLite 的完整流程提供可直接运行的实战代码并对关键技术与工程亮点进行深度解析。二、SQLite 核心特性与定位2.1 六大核心优势零配置、零依赖Python 标准库内置import sqlite3即可使用无需pip install真正做到开箱即用。单文件存储整个数据库表结构、索引、数据全部存放在一个.db文件中可随意拷贝、压缩、传输跨平台兼容Windows/macOS/Linux/Android/iOS。支持标准 SQL支持SELECT、JOIN、子查询、聚合函数、视图、触发器、外键需手动启用等SQL 技能可无缝迁移至 MySQL/PostgreSQL。ACID 事务保障完整支持原子性、一致性、隔离性、持久性确保数据安全。内存数据库模式支持:memory:模式数据库完全运行在内存中速度极快适合单元测试和临时数据处理。极低的资源占用库体积小几百 KB运行时内存开销低适合嵌入式设备和资源受限环境。2.2 适用与不适用场景维度适用场景不适用场景应用类型桌面工具、CLI 脚本、原型开发、单元测试高并发 Web 后端、分布式系统数据规模GB 级以内TB 级海量数据并发模式多读少写、单用户多用户同时高频写入部署环境单机、离线、嵌入式需要远程访问、主从复制三、Python 操作 SQLite 完整实战以下是一个完整的实战项目——学生信息管理系统涵盖建库建表、增删改查、事务控制、参数化防注入、字典式结果读取、WAL 并发优化等核心功能。3.1 完整代码 SQLite 学生信息管理系统 —— 完整实战示例 涵盖建库建表、CRUD、事务、参数化查询、row_factory、WAL 模式、异常处理 importsqlite3importosfromdatetimeimportdatetime DB_FILEstudents.dbdefget_connection(db_path:strDB_FILE)-sqlite3.Connection: 创建数据库连接并做关键优化配置。 亮点 1. 开启 WAL 模式 —— 读写并发大幅缓解 database is locked 错误 2. 设置 row_factory sqlite3.Row —— 查询结果可按列名访问类似字典 3. 设置 timeout —— 遇到锁等待时自动重试避免立即报错 connsqlite3.connect(db_path,timeout10)# 开启 WALWrite-Ahead Logging模式读不阻塞写写不阻塞读conn.execute(PRAGMA journal_modeWAL)# 设置结果行为字典支持 row[name] 按列名取值conn.row_factorysqlite3.Rowreturnconndefinit_db(conn:sqlite3.Connection)-None:初始化数据库表结构使用 IF NOT EXISTS 保证幂等性。cursorconn.cursor()cursor.execute( CREATE TABLE IF NOT EXISTS students ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER NOT NULL CHECK(age 0 AND age 150), gender TEXT DEFAULT M CHECK(gender IN (M, F)), major TEXT, gpa REAL CHECK(gpa 0 AND gpa 4.0), enrolled_date TEXT DEFAULT (datetime(now)), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) )# 为常用查询字段创建索引提升查询性能cursor.execute(CREATE INDEX IF NOT EXISTS idx_students_name ON students(name))cursor.execute(CREATE INDEX IF NOT EXISTS idx_students_major ON students(major))conn.commit()print([OK] 数据库表初始化完成。)defadd_student(conn:sqlite3.Connection,name:str,age:int,gender:strM,major:strNone,gpa:floatNone)-int: 插入一条学生记录。 亮点使用参数化查询? 占位符彻底杜绝 SQL 注入风险。 返回值新插入记录的自增 ID。 cursorconn.cursor()cursor.execute(INSERT INTO students (name, age, gender, major, gpa) VALUES (?, ?, ?, ?, ?),(name,age,gender,major,gpa))conn.commit()student_idcursor.lastrowidprint(f[OK] 已添加学生:{name}(ID{student_id}))returnstudent_iddefadd_students_batch(conn:sqlite3.Connection,students:list)-None: 批量插入学生记录。 亮点使用 executemany() 替代循环 execute() 减少 SQL 解析开销批量插入性能提升显著。 cursorconn.cursor()cursor.executemany(INSERT INTO students (name, age, gender, major, gpa) VALUES (?, ?, ?, ?, ?),students)conn.commit()print(f[OK] 批量添加{len(students)}条学生记录。)defquery_students(conn:sqlite3.Connection,major:strNone,min_gpa:floatNone)-list: 条件查询学生。 亮点 1. 参数化查询防注入 2. row_factory 使结果支持 row[name] 字典式访问 cursorconn.cursor()sqlSELECT id, name, age, gender, major, gpa, enrolled_date FROM students WHERE 11params[]ifmajor:sql AND major ?params.append(major)ifmin_gpaisnotNone:sql AND gpa ?params.append(min_gpa)sql ORDER BY gpa DESCcursor.execute(sql,tuple(params))rowscursor.fetchall()# 将 sqlite3.Row 对象转为字典列表方便后续处理或 JSON 序列化results[dict(row)forrowinrows]returnresultsdefupdate_student_gpa(conn:sqlite3.Connection,student_id:int,new_gpa:float)-bool:更新指定学生的 GPA。cursorconn.cursor()cursor.execute(UPDATE students SET gpa ? WHERE id ?,(new_gpa,student_id))conn.commit()ifcursor.rowcount0:print(f[OK] 学生 ID{student_id}的 GPA 已更新为{new_gpa}。)returnTrueelse:print(f[WARN] 未找到 ID{student_id}的学生。)returnFalsedefdelete_student(conn:sqlite3.Connection,student_id:int)-bool:删除指定学生记录。cursorconn.cursor()cursor.execute(DELETE FROM students WHERE id ?,(student_id,))conn.commit()ifcursor.rowcount0:print(f[OK] 学生 ID{student_id}已删除。)returnTrueelse:print(f[WARN] 未找到 ID{student_id}的学生。)returnFalsedeftransfer_students(conn:sqlite3.Connection,from_major:str,to_major:str)-int: 事务操作示例将某专业的所有学生转到另一个专业。 亮点使用 with conn 管理事务边界—— 正常结束自动 commit抛出异常自动 rollback 保证多条 SQL 的原子性要么全成功要么全回滚。 try:withconn:# 事务开始cursorconn.cursor()cursor.execute(UPDATE students SET major ? WHERE major ?,(to_major,from_major))affectedcursor.rowcountprint(f[OK] 事务提交{affected}名学生从{from_major}转到{to_major}。)returnaffectedexceptsqlite3.Errorase:print(f[ERROR] 事务回滚{e})return0defprint_all_students(conn:sqlite3.Connection)-None:格式化打印所有学生信息。studentsquery_students(conn)ifnotstudents:print( 无数据)returnprint(f\n{ID:5}{姓名:10}{年龄:5}{性别:5}{专业:15}{GPA:6}{入学日期})print(-*70)forsinstudents:print(f{s[id]:5}{s[name]:10}{s[age]:5}{s[gender]:5}f{s[major]or未分配:15}{s[gpa]orN/A:6}{s[enrolled_date]})print(f\n共{len(students)}条记录。\n)defget_db_size(db_path:strDB_FILE)-str:获取数据库文件大小。ifos.path.exists(db_path):size_bytesos.path.getsize(db_path)ifsize_bytes1024:returnf{size_bytes}Belifsize_bytes1024*1024:returnf{size_bytes/1024:.1f}KBelse:returnf{size_bytes/(1024*1024):.1f}MBreturn0 B# 主流程演示 if__name____main__:print(*50)print( SQLite 学生信息管理系统 —— 演示)print(*50)# 1. 连接数据库自动创建文件connget_connection()print(f[OK] 已连接数据库:{DB_FILE}(大小:{get_db_size()}))# 2. 初始化表结构init_db(conn)# 3. 插入单条记录add_student(conn,张三,20,M,计算机科学,3.8)add_student(conn,李四,21,F,数学,3.5)add_student(conn,王五,19,M,计算机科学,3.9)# 4. 批量插入batch_data[(赵六,22,F,物理,3.2),(孙七,20,M,数学,3.7),(周八,21,F,计算机科学,3.6),(吴九,19,M,物理,3.4),]add_students_batch(conn,batch_data)# 5. 查询全部print(\n--- 全部学生 ---)print_all_students(conn)# 6. 条件查询计算机专业且 GPA 3.5print(--- 计算机专业 GPA 3.5 ---)cs_studentsquery_students(conn,major计算机科学,min_gpa3.5)forsincs_students:print(f{s[name]}(GPA{s[gpa]}))# 7. 更新操作update_student_gpa(conn,student_id1,new_gpa3.95)# 8. 事务操作将物理专业学生转到应用物理transfer_students(conn,from_major物理,to_major应用物理)# 9. 删除操作delete_student(conn,student_id4)# 10. 最终状态print(\n--- 最终学生列表 ---)print_all_students(conn)# 11. 统计信息cursorconn.cursor()cursor.execute(SELECT COUNT(*) as total, AVG(gpa) as avg_gpa FROM students)statsdict(cursor.fetchone())print(f统计: 总人数{stats[total]}, 平均GPA{stats[avg_gpa]:.2f})print(f数据库文件大小:{get_db_size()})# 12. 关闭连接conn.close()print(\n[OK] 数据库连接已关闭。)3.2 运行结果示例 SQLite 学生信息管理系统 —— 演示 [OK] 已连接数据库: students.db (大小: 0 B) [OK] 数据库表初始化完成。 [OK] 已添加学生: 张三 (ID1) [OK] 已添加学生: 李四 (ID2) [OK] 已添加学生: 王五 (ID3) [OK] 批量添加 4 条学生记录。 --- 全部学生 --- ID 姓名 年龄 性别 专业 GPA 入学日期 ---------------------------------------------------------------------- 1 张三 20 M 计算机科学 3.8 2026-10-19 10:00:00 3 王五 19 M 计算机科学 3.9 2026-10-19 10:00:00 6 周八 21 F 计算机科学 3.6 2026-10-19 10:00:00 2 李四 21 F 数学 3.5 2026-10-19 10:00:00 5 孙七 20 M 数学 3.7 2026-10-19 10:00:00 4 赵六 22 F 物理 3.2 2026-10-19 10:00:00 7 吴九 19 M 物理 3.4 2026-10-19 10:00:00 共 7 条记录。 --- 计算机专业 GPA 3.5 --- 王五 (GPA3.9) 张三 (GPA3.8) 周八 (GPA3.6) [OK] 学生 ID1 的 GPA 已更新为 3.95。 [OK] 事务提交2 名学生从 物理 转到 应用物理。 [OK] 学生 ID4 已删除。 --- 最终学生列表 --- ... 统计: 总人数6, 平均GPA3.64 数据库文件大小: 16.0 KB [OK] 数据库连接已关闭。四、核心技术解析4.1 参数化查询——杜绝 SQL 注入的安全红线SQL 注入是最常见的 Web 安全漏洞之一。新手常犯的错误是用 f-string 或%拼接用户输入# 危险绝对不要这样写nameinput(输入姓名: )cursor.execute(fSELECT * FROM students WHERE name {name})如果用户输入 OR 11整个 WHERE 条件将被绕过全表数据泄露若输入; DROP TABLE students; --甚至可以直接删表。正确做法是使用?占位符的参数化查询cursor.execute(SELECT * FROM students WHERE name ?,(name,))数据库驱动会将参数值作为纯数据处理永远不会被解释为 SQL 语句的一部分。注意(name,)中的逗号——这是构造单元素元组的语法漏掉逗号会变成普通括号导致类型错误。4.2 事务管理——with conn的原子性保障SQLite 默认开启事务模式。执行INSERT/UPDATE/DELETE后如果不显式调用conn.commit()数据仅存在于未提交的事务中连接关闭时会被自动回滚丢弃。很多新手的困惑写了数据重启后没了根源就在于忘记commit()。Python 提供了优雅的事务管理方式——with conn上下文管理器withconn:cursor.execute(UPDATE accounts SET balance balance - 100 WHERE id 1)cursor.execute(UPDATE accounts SET balance balance 100 WHERE id 2)# 正常结束 → 自动 commit中途抛异常 → 自动 rollbackwith conn管理的是事务边界而非连接生命周期conn仍需另外close()。它保证了这一组操作的原子性——要么全部成功提交要么整体回滚非常适合转账、库存扣减等需要一致性的业务场景。4.3row_factory sqlite3.Row——字典式结果访问默认情况下fetchall()返回的是元组列表取值只能靠下标row[0]、row[1]列一多就容易数错后续加列还会导致下标全部错位。设置conn.row_factory sqlite3.Row后每行结果支持按列名访问conn.row_factorysqlite3.Row cursor.execute(SELECT id, name, gpa FROM students)rowcursor.fetchone()print(row[name])# 按列名取清晰不易错print(dict(row))# {id: 1, name: 张三, gpa: 3.95}sqlite3.Row几乎零性能开销同时兼容下标访问且dict(row)可直接序列化为 JSON是强烈推荐的最佳实践。4.4 WAL 模式——缓解并发锁冲突SQLite 默认的日志模式rollback journal下写操作会锁定整个数据库文件同一时刻只能有一个写者。多进程/多线程场景下极易触发database is locked错误。开启 WALWrite-Ahead Logging模式后读和写可以并发执行读不阻塞写、写不阻塞读大幅缓解锁冲突conn.execute(PRAGMA journal_modeWAL)WAL 是数据库级别的持久设置设置一次后一直有效会在同目录下生成.db-wal和.db-shm辅助文件。配合timeout参数可以让程序在遇到锁时自动等待重试而非立即报错。4.5executemany()——批量插入的性能优化循环调用execute()逐条插入时每条 SQL 都需要经过解析、编译、执行三个步骤。使用executemany()批量插入SQL 只需编译一次性能提升显著# 低效循环逐条插入forstudentinstudents:cursor.execute(INSERT INTO students (...) VALUES (...),student)# 高效批量插入cursor.executemany(INSERT INTO students (...) VALUES (...),students)对于千条级别的数据插入executemany()通常能带来数倍的性能提升。若追求极致性能还可以结合conn.execute(BEGIN)手动控制事务避免每条插入都隐式提交。