SQLite 完全指南

官方文档：https://www.sqlite.org/docs.html
适用版本：SQLite 3.43+（2026-05-07 整理）

目录

• 核心特点
• Python sqlite3 标准库
• Connection 对象
• Cursor 对象
• 参数绑定
• Row 对象
• 上下文管理器
• WAL 模式
• 常用 PRAGMA
• 内存数据库
• SQLAlchemy 集成
• 踩坑与注意事项

核心特点

SQLite 是一个无服务器、单文件的关系型数据库引擎，整个数据库（包括表、索引、数据）存储在一个 .db 文件中。

适用场景

• 嵌入式应用（桌面软件、移动端）
• 单元测试和集成测试（替代真实数据库）
• 小型工具和脚本（配置存储、本地缓存）
• 原型开发和快速验证
• 数据量在数 GB 以内的单机应用

与 MySQL/PostgreSQL 的差异

Python sqlite3 标准库

sqlite3 是 Python 标准库模块，无需安装，遵循 DB-API 2.0 规范（PEP 249）。

import sqlite3

sqlite3.connect()

conn = sqlite3.connect(
    database,
    timeout=5.0,
    detect_types=0,
    isolation_level='',
    check_same_thread=True
)

# 基础连接
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 对象，代表一个数据库连接。

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 语句和获取结果的核心对象。

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 注入。

? 位置占位符

# 单个参数
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 命名占位符

# 使用字典传参，可读性更好
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 对象，支持按列名访问。

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 直接返回字典：

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()

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：

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 模式

开启 WAL 模式

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 内容合并回主数据库文件：

# 手动触发 checkpoint
conn.execute('PRAGMA wal_checkpoint(TRUNCATE)')

SQLite 会在适当时机自动 checkpoint（WAL 文件超过 1000 页时），通常不需要手动触发。

常用 PRAGMA

PRAGMA 是 SQLite 的配置指令，用于调整运行时行为。

# 读取 PRAGMA 值
result = conn.execute('PRAGMA cache_size').fetchone()
print(result[0])

# 设置 PRAGMA 值
conn.execute('PRAGMA cache_size = -64000')  # 单位 KB（负数）

常用 PRAGMA 参数表

# 推荐的生产配置（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: 作为数据库路径，创建完全在内存中的数据库，进程退出后数据消失。

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 格式：

# 使用 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 访问同一内存数据库

内存数据库与磁盘数据库互导

# 从磁盘加载到内存（加速读取）
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 集成

同步连接字符串

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

pip install aiosqlite

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）：

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 集成示例

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 时，跨线程使用同一连接直接报错。

如果需要多线程写入，有两种方案：

方案一：每个线程独立连接（推荐）

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',))

方案二：共享连接 + 互斥锁

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 中通常这样处理：

engine = create_engine(
    'sqlite:///app.db',
    connect_args={'check_same_thread': False}
)

SQLAlchemy 的连接池本身确保了每次请求使用独立连接，因此此处是安全的。

数值类型亲和性（动态类型）

SQLite 使用动态类型系统，列类型只是"亲和性（affinity）"提示，实际存储类型由值决定：

# 坑：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 之前尤为严重）：

需要复杂的表结构变更时，通用方案是重建表：

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 外键约束默认关闭，需要每次连接时手动开启：

conn = sqlite3.connect('app.db')
conn.execute('PRAGMA foreign_keys=ON')

这个设置不持久化，每次建立新连接都要重新执行。可以用 event 监听器自动设置（SQLAlchemy）：

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()

大量写入时的性能优化

# 单条插入（慢，每条是一个事务）
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。

解决： 始终用参数化查询（? 占位符），让驱动负责转义。

cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

参见

• MySQL基础完全指南
• PostgreSQL完全指南
• 数据库设计规范