> ## Content Index > Fetch the complete content index at: https://blog.vercanti.com/llms.txt > Use this file to discover other available public pages before exploring further. # SQLite 完全指南 - URL: https://blog.vercanti.com/sqlite-wan-quan-zhi-nan/ - Published: 2026-08-28T14:35:38.000Z - Updated: 2026-08-28T14:59:16.000Z - Description: SQLite 是一个无服务器、单文件的关系型数据库引擎，整个数据库（包括表、索引、数据）存储在一个 .db 文件中。 sqlite3 是 Python 标准库模块，无需安装，遵循 DB-API 2.0 规范（PEP 249）。 sqlite3.connect() 返回 Connection 对象，代表一个数据库连接。 Cursor 是执行 SQL 语句和获取结果的核心对象。 始终使用参数绑定，严禁用字符串拼接构造 SQL，防止 SQL 注入。 Python 3.12 新增，与命名参数等价，此处不展开。 默认情况下，查询结果的每行是元组。设置 row_fa - Author: yellowdog - Tags: 数据库 > 官方文档： > 适用版本：SQLite 3.43+（2026-05-07 整理） --- ## 目录 - [核心特点](#%E6%A0%B8%E5%BF%83%E7%89%B9%E7%82%B9) - [Python sqlite3 标准库](#python-sqlite3-%E6%A0%87%E5%87%86%E5%BA%93) - [Connection 对象](#connection-%E5%AF%B9%E8%B1%A1) - [Cursor 对象](#cursor-%E5%AF%B9%E8%B1%A1) - [参数绑定](#%E5%8F%82%E6%95%B0%E7%BB%91%E5%AE%9A) - [Row 对象](#row-%E5%AF%B9%E8%B1%A1) - [上下文管理器](#%E4%B8%8A%E4%B8%8B%E6%96%87%E7%AE%A1%E7%90%86%E5%99%A8) - [WAL 模式](#wal-%E6%A8%A1%E5%BC%8F) - [常用 PRAGMA](#%E5%B8%B8%E7%94%A8-pragma) - [内存数据库](#%E5%86%85%E5%AD%98%E6%95%B0%E6%8D%AE%E5%BA%93) - [SQLAlchemy 集成](#sqlalchemy-%E9%9B%86%E6%88%90) - [踩坑与注意事项](#%E8%B8%A9%E5%9D%91%E4%B8%8E%E6%B3%A8%E6%84%8F%E4%BA%8B%E9%A1%B9) --- ## 核心特点 SQLite 是一个无服务器、单文件的关系型数据库引擎，整个数据库（包括表、索引、数据）存储在一个 `.db` 文件中。 ### 适用场景 - 嵌入式应用（桌面软件、移动端） - 单元测试和集成测试（替代真实数据库） - 小型工具和脚本（配置存储、本地缓存） - 原型开发和快速验证 - 数据量在数 GB 以内的单机应用 ### 与 MySQL/PostgreSQL 的差异 | 特性 | SQLite | MySQL | PostgreSQL | | ----------- | ------------- | ----------- | ---------- | | 架构 | 无服务器，进程内库 | 客户端/服务器 | 客户端/服务器 | | 存储 | 单文件 | 多文件目录 | 数据目录 | | 并发写入 | 同一时刻只有一个写入者 | 行级锁，高并发 | MVCC，高并发 | | 并发读取 | WAL 模式下支持多读单写 | 支持 | 支持 | | 数据类型 | 动态类型（类型亲和性） | 强类型 | 强类型 | | ALTER TABLE | 功能有限（不能删列/改列） | 完整支持 | 完整支持 | | 外键 | 支持但默认关闭 | 支持（InnoDB） | 支持 | | JSON | 通过扩展支持 | 原生 JSON 类型 | 原生 JSONB | | 全文搜索 | FTS5 扩展 | FULLTEXT 索引 | tsvector | | 网络访问 | 不支持 | 支持 | 支持 | | 最大数据库大小 | 281 TB（理论） | 无限制 | 无限制 | | 适用场景 | 嵌入式/测试/小工具 | 中大型 Web 应用 | 复杂查询/大型应用 | --- ## Python sqlite3 标准库 `sqlite3` 是 Python 标准库模块，无需安装，遵循 DB-API 2.0 规范（PEP 249）。 ```python import sqlite3 ``` ### sqlite3.connect() ```python conn = sqlite3.connect( database, timeout=5.0, detect_types=0, isolation_level='', check_same_thread=True ) ``` | 参数 | 类型 | 默认值 | 说明 | | ------------------- | -------- | ---- | ----------------------------------------------------------------------------------------- | | database | str | 必填 | 数据库文件路径；:memory: 为内存数据库；'' 为临时文件数据库 | | timeout | float | 5.0 | 等待锁释放的超时时间（秒），超时后抛出 OperationalError | | detect\_types | int | 0 | 类型检测模式：sqlite3.PARSE\_DECLTYPES 按列声明类型转换，sqlite3.PARSE\_COLNAMES 按列别名转换，可用 \` | | isolation\_level | str/None | '' | 事务隔离级别：'' 自动提交 DDL，DML 开启隐式事务；None 关闭隐式事务（手动控制）；'DEFERRED'/'IMMEDIATE'/'EXCLUSIVE' 显式事务类型 | | check\_same\_thread | bool | True | 是否检查跨线程使用，多线程时需设为 False（见踩坑） | ```python # 基础连接 conn = sqlite3.connect('app.db') # 关闭隐式事务（完全手动控制） conn = sqlite3.connect('app.db', isolation_level=None) # 启用类型检测（配合 register_adapter/register_converter 使用） conn = sqlite3.connect( 'app.db', detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES ) ``` --- ## Connection 对象 `sqlite3.connect()` 返回 `Connection` 对象，代表一个数据库连接。 | 方法/属性 | 说明 | | ---------------------------------- | -------------------------------- | | cursor() | 创建并返回 Cursor 对象 | | commit() | 提交当前事务 | | rollback() | 回滚当前事务 | | close() | 关闭连接（未提交的事务会回滚） | | execute(sql, params) | 快捷方式，内部创建 Cursor 执行，返回 Cursor | | executemany(sql, seq) | 快捷方式，批量执行 | | executescript(sql\_script) | 执行多条 SQL 语句（以 ; 分隔的脚本），自动提交未提交事务 | | create\_function(name, narg, func) | 注册自定义 SQL 函数 | | create\_aggregate(name, narg, cls) | 注册自定义聚合函数 | | row\_factory | 属性，设置行工厂（如 sqlite3.Row） | | text\_factory | 属性，设置文本解码方式，默认 str | | isolation\_level | 属性，读取或修改隔离级别 | | in\_transaction | 属性（只读），当前是否在事务中 | | total\_changes | 属性，连接生命周期内的总变更行数 | ```python conn = sqlite3.connect('app.db') # 创建表 conn.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER ) ''') conn.commit() # 插入数据 conn.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30)) conn.commit() # 批量插入 conn.executemany( 'INSERT INTO users (name, age) VALUES (?, ?)', [('Bob', 25), ('Carol', 28)] ) conn.commit() # 注册自定义函数 import hashlib conn.create_function('md5', 1, lambda s: hashlib.md5(s.encode()).hexdigest()) rows = conn.execute("SELECT md5(name) FROM users").fetchall() conn.close() ``` --- ## Cursor 对象 `Cursor` 是执行 SQL 语句和获取结果的核心对象。 | 方法/属性 | 说明 | | ------------------------------------- | ---------------------------------------------------- | | execute(sql, parameters=()) | 执行单条 SQL 语句 | | executemany(sql, seq\_of\_parameters) | 对序列中每组参数执行 SQL（仅 INSERT/UPDATE/DELETE） | | executescript(sql\_script) | 执行 SQL 脚本（多条语句） | | fetchone() | 返回下一行，无更多行时返回 None | | fetchall() | 返回所有剩余行，结果为列表 | | fetchmany(size=cursor.arraysize) | 返回最多 size 行，默认 arraysize（默认值 1） | | rowcount | 上次 execute() 影响的行数（INSERT/UPDATE/DELETE），SELECT 为 -1 | | lastrowid | 上次 INSERT 操作的行 ID（仅 execute() 有效） | | description | 查询结果的列描述（7 元组列表），SELECT 后有值 | | arraysize | fetchmany() 默认大小，默认 1，可修改 | | close() | 关闭游标 | ```python conn = sqlite3.connect('app.db') cursor = conn.cursor() # execute + fetchall cursor.execute('SELECT id, name, age FROM users WHERE age > ?', (20,)) rows = cursor.fetchall() # [(1, 'Alice', 30), (2, 'Bob', 25), ...] # execute + fetchone（逐行） cursor.execute('SELECT * FROM users ORDER BY id') while True: row = cursor.fetchone() if row is None: break print(row) # execute + fetchmany（分批处理大结果集） cursor.execute('SELECT * FROM users') cursor.arraysize = 500 while True: batch = cursor.fetchmany() if not batch: break process(batch) # executemany（批量写入） data = [('Dave', 22), ('Eve', 35)] cursor.executemany('INSERT INTO users (name, age) VALUES (?, ?)', data) print(cursor.rowcount) # 2 conn.commit() # lastrowid cursor.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Frank', 40)) print(cursor.lastrowid) # 新插入行的 id conn.commit() # description（列信息） cursor.execute('SELECT id, name FROM users LIMIT 1') print(cursor.description) # [('id', None, None, None, None, None, None), # ('name', None, None, None, None, None, None)] cursor.close() conn.close() ``` --- ## 参数绑定 始终使用参数绑定，严禁用字符串拼接构造 SQL，防止 SQL 注入。 ### `?` 位置占位符 ```python # 单个参数 cursor.execute('SELECT * FROM users WHERE id = ?', (1,)) # 注意：即使只有一个参数也必须传元组 # 多个参数 cursor.execute( 'SELECT * FROM users WHERE name = ? AND age > ?', ('Alice', 20) ) # executemany cursor.executemany( 'INSERT INTO users (name, age) VALUES (?, ?)', [('Alice', 30), ('Bob', 25)] ) ``` ### `:name` 命名占位符 ```python # 使用字典传参，可读性更好 cursor.execute( 'SELECT * FROM users WHERE name = :name AND age > :min_age', {'name': 'Alice', 'min_age': 20} ) # executemany 也支持命名参数 cursor.executemany( 'INSERT INTO users (name, age) VALUES (:name, :age)', [{'name': 'Alice', 'age': 30}, {'name': 'Bob', 'age': 25}] ) ``` ### `$name` 占位符（Python 3.12+） Python 3.12 新增，与命名参数等价，此处不展开。 --- ## Row 对象 默认情况下，查询结果的每行是元组。设置 `row_factory = sqlite3.Row` 后，每行变为 `Row` 对象，支持按列名访问。 ```python conn = sqlite3.connect('app.db') conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute('SELECT id, name, age FROM users WHERE id = 1') row = cursor.fetchone() # 按列名访问（推荐） print(row['name']) # 'Alice' print(row['age']) # 30 # 按索引访问（兼容元组语法） print(row[0]) # 1 # 转为字典 row_dict = dict(row) # {'id': 1, 'name': 'Alice', 'age': 30} # keys() 获取列名 print(row.keys()) # ['id', 'name', 'age'] conn.close() ``` 也可以自定义 `row_factory` 直接返回字典： ```python def dict_factory(cursor, row): return {col[0]: row[idx] for idx, col in enumerate(cursor.description)} conn.row_factory = dict_factory ``` --- ## 上下文管理器 `Connection` 对象支持上下文管理器协议（`with` 语句），但行为与直觉有区别： - `with conn:` **不负责关闭连接**，只管理事务 - 代码块正常结束 → 自动 `commit()` - 代码块抛出异常 → 自动 `rollback()` ```python conn = sqlite3.connect('app.db') # 事务管理 with conn: conn.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30)) conn.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Bob', 25)) # 正常结束，自动 commit # 回滚示例 try: with conn: conn.execute('INSERT INTO users (name) VALUES (?)', ('Carol',)) raise ValueError('故意报错') # 抛出异常，自动 rollback，Carol 不会被插入 except ValueError: pass # 连接需要手动关闭 conn.close() ``` 如果需要同时管理连接的生命周期，用 `contextlib.closing`： ```python from contextlib import closing import sqlite3 with closing(sqlite3.connect('app.db')) as conn: with conn: conn.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30)) # 退出外层 with 时自动关闭连接 ``` --- ## WAL 模式 WAL（Write-Ahead Logging，预写日志）是 SQLite 的一种日志模式，相比默认的 DELETE 模式有更好的并发读性能。 ### 默认模式 vs WAL 模式 | 特性 | DELETE 模式（默认） | WAL 模式 | | ----- | ------------- | ------------------------- | | 读写并发 | 读写互斥（写时不能读） | 读写不阻塞（写时可以读） | | 写写并发 | 只允许一个写入者 | 只允许一个写入者 | | 写入性能 | 一般 | 更好（顺序写入 WAL 文件） | | 读取性能 | 一般 | 稍有开销（需检查 WAL 文件） | | 崩溃恢复 | 快 | 快 | | 跨进程共享 | 支持 | 支持（同一文件系统） | | 文件数量 | 1 个 .db 文件 | 额外产生 .db-wal 和 .db-shm 文件 | ### 开启 WAL 模式 ```python import sqlite3 conn = sqlite3.connect('app.db') # 开启 WAL 模式（持久化，重连后生效） conn.execute('PRAGMA journal_mode=WAL') conn.commit() # 验证当前模式 result = conn.execute('PRAGMA journal_mode').fetchone() print(result[0]) # 'wal' ``` WAL 模式是持久化设置，设置一次后数据库文件记录该模式，后续连接无需重新设置（但建议每次连接都执行，以免切换到其他模式的数据库）。 ### WAL checkpoint WAL 文件会持续增长，需要定期 checkpoint 将 WAL 内容合并回主数据库文件： ```python # 手动触发 checkpoint conn.execute('PRAGMA wal_checkpoint(TRUNCATE)') ``` SQLite 会在适当时机自动 checkpoint（WAL 文件超过 1000 页时），通常不需要手动触发。 --- ## 常用 PRAGMA PRAGMA 是 SQLite 的配置指令，用于调整运行时行为。 ```python # 读取 PRAGMA 值 result = conn.execute('PRAGMA cache_size').fetchone() print(result[0]) # 设置 PRAGMA 值 conn.execute('PRAGMA cache_size = -64000') # 单位 KB（负数） ``` ### 常用 PRAGMA 参数表 | PRAGMA | 默认值 | 说明 | | ------------------- | -------------------- | ------------------------------------------------ | | cache\_size | \-2000（约 2 MB） | 页缓存大小，负数为 KB，正数为页数，增大可提升读性能 | | synchronous | NORMAL（WAL 模式）或 FULL | 磁盘同步级别：OFF（最快，不安全）/NORMAL（推荐）/FULL（最安全，最慢）/EXTRA | | foreign\_keys | OFF | 外键约束：ON 开启，OFF 关闭（SQLite 默认不开启！） | | temp\_store | DEFAULT（0） | 临时表/索引存储位置：0\=默认/1\=文件/2\=内存 | | journal\_mode | DELETE | 日志模式：DELETE/WAL/MEMORY/OFF | | mmap\_size | 0 | 内存映射文件大小（字节），0 关闭 mmap，大文件读取可开启 | | page\_size | 4096 | 数据库页大小（只能在创建时设置），常用 4096 或 8192 | | auto\_vacuum | NONE | 自动整理碎片：NONE/FULL/INCREMENTAL | | busy\_timeout | 0 | 等待锁的超时毫秒数，等价于 connect(timeout=...) | | wal\_autocheckpoint | 1000 | WAL 模式下自动 checkpoint 的页数阈值 | ```python # 推荐的生产配置（WAL 模式） def configure_connection(conn: sqlite3.Connection) -> None: conn.execute('PRAGMA journal_mode=WAL') conn.execute('PRAGMA synchronous=NORMAL') conn.execute('PRAGMA foreign_keys=ON') conn.execute('PRAGMA cache_size=-64000') # 64 MB 缓存 conn.execute('PRAGMA temp_store=MEMORY') conn.execute('PRAGMA busy_timeout=5000') # 等待锁 5 秒 ``` --- ## 内存数据库 传入 `:memory:` 作为数据库路径，创建完全在内存中的数据库，进程退出后数据消失。 ```python import sqlite3 # 创建内存数据库 conn = sqlite3.connect(':memory:') conn.execute(''' CREATE TABLE users ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, age INTEGER ) ''') conn.execute("INSERT INTO users (name, age) VALUES ('Alice', 30)") conn.commit() result = conn.execute('SELECT * FROM users').fetchall() print(result) # [(1, 'Alice', 30)] conn.close() # 关闭后数据消失 ``` ### 内存数据库的适用场景 - 单元测试：测试 ORM 模型和查询逻辑，不需要清理文件 - 临时计算：将 CSV/JSON 数据加载进内存数据库后做 SQL 聚合 - 缓存层：应用启动时从磁盘读取热数据到内存数据库加速查询 ### 共享内存数据库（多连接访问同一内存库） 普通 `:memory:` 每次 `connect()` 都创建独立的内存库。要让多个连接访问同一内存库，使用 URI 格式： ```python # 使用 URI 创建共享内存数据库（需要 check_same_thread=False） conn1 = sqlite3.connect('file:shared_mem?mode=memory&cache=shared', uri=True) conn2 = sqlite3.connect('file:shared_mem?mode=memory&cache=shared', uri=True) # conn1 和 conn2 访问同一内存数据库 ``` ### 内存数据库与磁盘数据库互导 ```python # 从磁盘加载到内存（加速读取） disk_conn = sqlite3.connect('app.db') mem_conn = sqlite3.connect(':memory:') disk_conn.backup(mem_conn) disk_conn.close() # 后续操作在 mem_conn 上进行，速度更快 # 将内存数据库保存到磁盘 mem_conn = sqlite3.connect(':memory:') # ... 操作 mem_conn ... disk_conn = sqlite3.connect('result.db') mem_conn.backup(disk_conn) disk_conn.close() mem_conn.close() ``` --- ## SQLAlchemy 集成 ### 同步连接字符串 ```python from sqlalchemy import create_engine # 相对路径 engine = create_engine('sqlite:///app.db') # 绝对路径（Linux/macOS） engine = create_engine('sqlite:////home/user/app.db') # 绝对路径（Windows） engine = create_engine(r'sqlite:///C:\Users\user\app.db') # 内存数据库 engine = create_engine('sqlite://') # 每次连接新建，适合测试 engine = create_engine('sqlite:///:memory:') # 同上 # 共享内存数据库（多个连接访问同一内存库） engine = create_engine( 'sqlite:///:memory:', connect_args={'check_same_thread': False}, poolclass=StaticPool # 强制所有连接使用同一底层连接 ) ``` ### 异步：aiosqlite ```bash pip install aiosqlite ``` ```python from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker # aiosqlite 作为异步驱动 engine = create_async_engine('sqlite+aiosqlite:///app.db', echo=True) AsyncSessionLocal = sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False ) async def get_users(): async with AsyncSessionLocal() as session: result = await session.execute(select(User)) return result.scalars().all() ``` 直接使用 `aiosqlite`（不经过 SQLAlchemy）： ```python import aiosqlite import asyncio async def main(): async with aiosqlite.connect('app.db') as conn: conn.row_factory = aiosqlite.Row async with conn.execute('SELECT * FROM users') as cursor: async for row in cursor: print(dict(row)) await conn.execute( 'INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30) ) await conn.commit() asyncio.run(main()) ``` ### FastAPI 集成示例 ```python from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import AsyncSession from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app: FastAPI): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield await engine.dispose() app = FastAPI(lifespan=lifespan) async def get_db(): async with AsyncSessionLocal() as session: yield session @app.get('/users') async def list_users(db: AsyncSession = Depends(get_db)): result = await db.execute(select(User)) return result.scalars().all() ``` --- ## 踩坑与注意事项 ### 多线程写入需加锁 SQLite 同一时刻只允许一个写入者。默认设置 `check_same_thread=True` 时，跨线程使用同一连接直接报错。 如果需要多线程写入，有两种方案： **方案一：每个线程独立连接（推荐）** ```python import threading import sqlite3 _local = threading.local() def get_conn(): if not hasattr(_local, 'conn'): _local.conn = sqlite3.connect('app.db') _local.conn.execute('PRAGMA journal_mode=WAL') return _local.conn def worker(): conn = get_conn() # 每个线程自己的连接 with conn: conn.execute('INSERT INTO logs (msg) VALUES (?)', ('log',)) ``` **方案二：共享连接 + 互斥锁** ```python import sqlite3 import threading conn = sqlite3.connect('app.db', check_same_thread=False) _lock = threading.Lock() def write_log(msg: str): with _lock: with conn: conn.execute('INSERT INTO logs (msg) VALUES (?)', (msg,)) ``` ### check\_same\_thread=False 的风险 设置 `check_same_thread=False` 关闭了线程检查，但 SQLite 连接对象本身**不是线程安全的**。如果多个线程在不加锁的情况下并发操作同一个连接，会导致数据库损坏或程序崩溃。设置此参数时，必须配合外部锁机制。 FastAPI + SQLAlchemy 中通常这样处理： ```python engine = create_engine( 'sqlite:///app.db', connect_args={'check_same_thread': False} ) ``` SQLAlchemy 的连接池本身确保了每次请求使用独立连接，因此此处是安全的。 ### 数值类型亲和性（动态类型） SQLite 使用动态类型系统，列类型只是"亲和性（affinity）"提示，实际存储类型由值决定： | 类型亲和性 | 列声明中包含 | 实际存储 | | ------- | ---------------------- | ---------------- | | INTEGER | INT、INTEGER | 整数存为整数 | | REAL | REAL、FLOAT、DOUBLE | 浮点存为 8 字节浮点 | | TEXT | TEXT、CHAR、CLOB、VARCHAR | 文本存为 UTF-8/16 | | BLOB | BLOB 或无类型 | 按原始数据存储 | | NUMERIC | 其他声明 | 尝试转为整数或浮点，否则存为文本 | ```python # 坑：SQLite 接受任意类型，不报错 conn.execute("CREATE TABLE t (val INTEGER)") conn.execute("INSERT INTO t VALUES (?)", ('hello',)) # 不会报错！ row = conn.execute("SELECT val, typeof(val) FROM t").fetchone() print(row) # ('hello', 'text') ``` 如果需要严格类型检查，需在应用层验证（使用 Pydantic、SQLAlchemy 类型等）。 ### ALTER TABLE 功能限制 SQLite 对 `ALTER TABLE` 的支持非常有限（SQLite 3.35.0 之前尤为严重）： | 操作 | 支持情况 | | ------------- | ----------------------- | | ADD COLUMN | 支持（有限制：不能有默认值表达式、不能是主键） | | DROP COLUMN | SQLite 3.35.0+ 支持（有限制） | | RENAME COLUMN | SQLite 3.25.0+ 支持 | | RENAME TABLE | 支持 | | 修改列类型 | 不支持 | | 添加/删除约束 | 不支持 | 需要复杂的表结构变更时，通用方案是重建表： ```python with conn: conn.execute('CREATE TABLE users_new AS SELECT id, name FROM users') conn.execute('DROP TABLE users') conn.execute('ALTER TABLE users_new RENAME TO users') ``` ### 外键默认关闭 SQLite 外键约束默认关闭，需要每次连接时手动开启： ```python conn = sqlite3.connect('app.db') conn.execute('PRAGMA foreign_keys=ON') ``` 这个设置不持久化，每次建立新连接都要重新执行。可以用 `event` 监听器自动设置（SQLAlchemy）： ```python from sqlalchemy import event @event.listens_for(engine, 'connect') def set_sqlite_pragma(dbapi_conn, connection_record): cursor = dbapi_conn.cursor() cursor.execute('PRAGMA foreign_keys=ON') cursor.execute('PRAGMA journal_mode=WAL') cursor.close() ``` ### 大量写入时的性能优化 ```python # 单条插入（慢，每条是一个事务） for row in rows: conn.execute('INSERT INTO t VALUES (?)', (row,)) conn.commit() # 批量插入（快，一个事务提交所有） with conn: conn.executemany('INSERT INTO t VALUES (?)', [(row,) for row in rows]) # 更快：关闭同步写入（崩溃可能丢失数据，测试环境可用） conn.execute('PRAGMA synchronous=OFF') conn.execute('PRAGMA journal_mode=MEMORY') with conn: conn.executemany('INSERT INTO t VALUES (?)', data) ``` --- ## 最佳实践 **所有写操作必须在事务内执行，提升性能的同时保证原子性**：SQLite 的每条独立 `INSERT/UPDATE` 默认是独立事务（自动提交），磁盘 fsync 开销极大。用 `with conn:` 包裹批量操作，一次提交可提升写入速度 100 倍以上。 **使用 WAL 模式提升并发读性能**：`PRAGMA journal_mode=WAL` 允许多个读者与一个写者并发运行，读操作不阻塞写操作，适合 Web 应用的读多写少场景。 **为 WHERE、JOIN、ORDER BY 中的列创建索引**：SQLite 没有自动统计，不会主动优化全表扫描。用 `EXPLAIN QUERY PLAN` 确认查询是否命中索引，未命中则手动创建。 **生产环境不要在并发写入场景用 SQLite**：SQLite 写操作是文件级锁，高并发写入时性能急剧下降。轻量应用（单用户桌面、嵌入式）适合 SQLite；多用户 Web 应用应使用 PostgreSQL 或 MySQL。 --- ## 常见陷阱 ### 陷阱：并发写入时抛出 `database is locked` **现象：** 多线程或多进程同时写入 SQLite 时，频繁抛出 `sqlite3.OperationalError: database is locked`。 **原因：** SQLite 写操作持有独占文件锁，其他写操作必须等待。默认超时时间（`timeout`）很短，多线程争用时容易超时。 **解决：** 设置较长的 `timeout`（`connect(db, timeout=30)`），开启 WAL 模式，或将 SQLite 替换为支持并发的数据库。 ### 陷阱：不用事务导致批量写入极慢 **现象：** 插入 1 万行数据需要数十秒，而同等数据量在 MySQL 只需不到 1 秒。 **原因：** 默认情况下每条 `INSERT` 是一个独立事务，每次提交都需要 fsync，磁盘 IO 是瓶颈。 **解决：** 用 `with conn:` 上下文管理器将批量操作放在一个事务中。 ### 陷阱：直接格式化 SQL 字符串导致 SQL 注入 **现象：** 代码中出现 `f"SELECT * FROM users WHERE name = '{name}'"` 的写法，用户输入可以改变 SQL 结构。 **原因：** 字符串拼接不对特殊字符（`'`, `"`）转义，恶意输入可以闭合引号并注入任意 SQL。 **解决：** 始终用参数化查询（`?` 占位符），让驱动负责转义。 ```python cursor.execute("SELECT * FROM users WHERE name = ?", (name,)) ``` --- ## 参见 - [MySQL基础完全指南](https://blog.vercanti.com/mysql-ji-chu-wan-quan-zhi-nan/) - [PostgreSQL完全指南](https://blog.vercanti.com/postgresql-wan-quan-zhi-nan/) - [数据库设计规范](https://blog.vercanti.com/shu-ju-ku-she-ji-gui-fan/)