Python 与数据库

Python 应用几乎都要碰数据库。DB-API 是统一接口，SQLAlchemy 是上层抽象，但事务、索引、连接池最终还是数据库的事。把 Python 侧的调用和数据库侧的执行对上，才能解释「为什么加了索引还是慢」「为什么连接池会被打满」。

1. 一张总表

数据库概念	Python 侧	应用开发中的表现
连接	sqlite3.connect、psycopg.connect、SQLAlchemy Engine	连接泄漏、连接池耗尽
游标	cursor、Session.execute	结果集读取、流式处理
事务	commit/rollback、with conn、Session.begin	数据一致性、长事务、锁等待
SQL 参数	?、%s、$1、SQLAlchemy 表达式	SQL 注入、执行计划缓存
索引	CREATE INDEX、EXPLAIN	查询快慢、写入放大
连接池	psycopg_pool、SQLAlchemy QueuePool	并发容量、数据库 max_connections
ORM	SQLAlchemy、Django ORM	开发效率、N+1、懒加载
迁移	Alembic、Django migrations	schema 版本管理

2. DB-API：最小的共同接口

Python 的 DB-API 2.0 规定了连接、游标、参数风格和异常体系。SQLite 标准库就是最直接的例子：

import sqlite3

conn = sqlite3.connect("app.db")
conn.row_factory = sqlite3.Row          # 让查询结果支持 row["name"]

cur = conn.execute(
    "SELECT id, name FROM users WHERE email = ?",
    ("alice@example.com",),             # 参数化，不要拼字符串
)
row = cur.fetchone()
print(dict(row) if row else None)

conn.close()

不同驱动的占位符不一样：

驱动	占位符
sqlite3	? 或 :name
psycopg / psycopg2	%s 或 %(name)s
asyncpg	$1、$2
SQLAlchemy	由方言生成，业务侧用表达式

2.1 永远不要拼接 SQL

# 片段：conn / cur 已创建
# 错误：SQL 注入
email = "alice@example.com"      # 实际来自请求参数
cur.execute(f"SELECT * FROM users WHERE email = '{email}'")

# 正确：参数化
cur.execute("SELECT * FROM users WHERE email = ?", (email,))

参数化不仅防注入，也可能让数据库复用执行计划（是否复用取决于驱动和数据库，例如 MySQL 需要 useServerPrepStmts=true）。表名、列名、ORDER BY 字段不能用占位符，必须用白名单校验后拼进 SQL。

2.2 大数据量不要 fetchall

fetchall() 会把整个结果集读进内存。数据量大时用游标迭代或 fetchmany，但不同驱动行为不同：

cur = conn.execute("SELECT id, payload FROM events")
while rows := cur.fetchmany(1000):
    for row in rows:
        process(row)

• SQLite：fetchmany 就够，游标不会一次性物化全部行。
• psycopg2：普通 cursor 在 execute() 时已经把结果收进客户端，fetchmany 只是分批取 Python 对象；要真正流式，用 named cursor（conn.cursor(name="srv") + itersize）。
• psycopg3：用 server-side cursor。
• asyncpg：fetch() 一次性返回全部，分批要用 conn.cursor()。

需要严格流式时，先确认驱动的默认行为，不要假设 fetchmany 一定降低了内存峰值。

3. 事务：一致性边界

3.1 with conn 提交的是事务，不是关闭连接

在 sqlite3 / psycopg2 里，with conn 是事务上下文：成功则 commit，异常则 rollback，但连接仍然打开。

with conn:                       # 成功则 commit，异常则 rollback
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
# 这里连接仍然是打开的
conn.close()

SQLAlchemy 的 Session 同理：

from sqlalchemy.orm import Session

# 片段：engine / Order 省略
with Session(engine) as session:
    with session.begin():
        session.add(Order(user_id=1, amount=99))

with Session(...) 关闭的是 session；with session.begin() 管理的是事务。两层不要混淆。

3.2 ACID 与隔离级别

• 原子性：事务内的操作要么全成功，要么全回滚。
• 一致性：约束、外键、唯一索引在事务结束时成立。
• 隔离性：并发事务之间看到什么，由隔离级别决定。
• 持久性：提交后的数据落到持久存储。

常见隔离级别与现象：

级别	脏读	不可重复读	幻读
READ UNCOMMITTED	可能	可能	可能
READ COMMITTED	否	可能	可能
REPEATABLE READ	否	否	可能（取决于实现）
SERIALIZABLE	否	否	否

PostgreSQL 默认 READ COMMITTED；MySQL InnoDB 默认 REPEATABLE READ；SQLite 引擎默认是 serializable。但 Python sqlite3 的默认事务行为和引擎不是一回事：

• 默认 isolation_level=''（legacy 模式）下，SELECT 不会开启事务，只有 DML 才会隐式 BEGIN；
• with conn: 里的多条 SELECT 不保证看到同一快照；
• Python 3.12+ 可以用 autocommit=False 显式控制，或手动 BEGIN；
• 需要读快照时，显式开启事务，不要依赖引擎的默认隔离级别。

隔离级别越高，并发冲突和锁等待越多。

3.3 长事务是事故源

一个事务开着不提交，会：

• 持有行锁/间隙锁，阻塞其他写入；
• 让 MVCC 的旧版本无法回收，表膨胀；
• 让连接一直占用，池子被耗尽。

典型错误是把「读数据 → 调外部 API → 写数据」放在一个事务里。外部 API 可能超时几十秒，数据库连接和锁就跟着挂几十秒。正确做法是缩小事务范围：先读、提交，调用外部服务，再用短事务写入。

4. 索引：为什么加了还是慢

4.1 B+ 树不是 Python 的 dict

	Python dict	数据库索引
存储	内存哈希表	磁盘 B+ 树（或内存中的页）
查找	平均 O(1)	O(log n)，但受磁盘 I/O 影响
范围查询	不支持	天然支持 BETWEEN、ORDER BY
排序	无	索引顺序可直接满足排序
成本	内存占用	写入放大、页分裂、维护成本

所以「Python 里 dict 查得快」不能推出「数据库加索引就一定快」；索引是否被用上，要看查询条件、选择性和执行计划。

4.2 用 EXPLAIN 看执行计划

SQLite：

for row in conn.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ("a@b.com",)):
    print(row)

PostgreSQL：

EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'a@b.com';

关注：

• SCAN table 还是 SEARCH table USING INDEX；
• 预估行数 vs 实际行数是否偏差很大；
• 有没有额外的排序、临时表、回表。

4.3 联合索引的最左前缀

索引 (user_id, created_at) 可以支持：

WHERE user_id = 1
WHERE user_id = 1 ORDER BY created_at
WHERE user_id = 1 AND created_at > '2026-01-01'

但不能单独支持 WHERE created_at > ...，因为 created_at 不是最左列。设计联合索引时，把等值条件放前面、范围/排序放后面。

4.4 索引失效的常见形态

• 在索引列上做函数或运算：WHERE DATE(created_at) = '2026-01-01'；
• 隐式类型转换：字符串列和数字比较；
• 前导通配符：LIKE '%abc'；
• OR 连接了未建索引的列；
• 选择性太差：性别、状态这种只有几个值的列，单列索引往往没用；
• 统计信息过期：需要 ANALYZE。

5. 连接池：并发容量的闸门

每次请求都新建数据库连接，代价很高：TCP 握手、认证、会话初始化。连接池复用连接，但池子大小必须和数据库容量对齐。

SQLAlchemy：

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg://user:pass@localhost/app",
    pool_size=5,             # 常驻连接
    max_overflow=10,         # 峰值可临时超出
    pool_timeout=30,         # 等连接的超时
    pool_pre_ping=True,      # 取连接前先探活
    pool_recycle=1800,       # 定期回收，避免被中间件断开
)

psycopg_pool：

from psycopg_pool import ConnectionPool

with ConnectionPool(conninfo="postgresql://...", min_size=2, max_size=10) as pool:
    with pool.connection() as conn:
        ...

pool.open() 也可以显式打开；3.2+ 未打开就 connection() 可能隐式打开并告警。

几个硬约束：

• 所有进程的 pool_size + max_overflow 总和不能超过数据库 max_connections（还要给运维、迁移留余量）。
• Web 服务用多 worker 时，每个 worker 都有自己的池；峰值是 workers × (pool_size + max_overflow)，不是 workers × pool_size。
• 连接池不是越大越好：数据库的活跃连接太多，上下文切换和锁竞争会让整体更慢。
• 中间件（PgBouncer）能复用连接，但 transaction pooling 模式下不支持会话级状态、预处理语句要小心。

6. ORM：效率与陷阱

SQLAlchemy 2.0 的查询风格：

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    stmt = select(User).where(User.email == "alice@example.com")
    user = session.execute(stmt).scalar_one_or_none()

6.1 N+1 查询

# 每个 user 都触发一次 orders 查询：1 + N
for user in session.execute(select(User)).scalars():
    print(user.orders)

用 selectinload（一条额外查询）或 joinedload（JOIN）预加载：

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.orders))

• 一对多、多对多：优先 selectinload，避免 JOIN 产生笛卡尔积；
• 多对一：joinedload 通常更合适；
• 只读场景可以用 load_only 限制列。

6.2 Session 的生命周期

Web 应用里常见「每个请求一个 Session」：

from fastapi import Depends

def get_session():
    with Session(engine) as session:
        yield session

@app.get("/users/{user_id}")
def get_user(user_id: int, session: Session = Depends(get_session)):
    return session.get(User, user_id)

要点：

• 请求结束一定要关闭 session，否则连接不还池；
• 不要在 session 里做长时间的外部调用；
• 异步应用用 AsyncSession + 异步驱动，别在 async 端点里用同步 session。

6.3 迁移

schema 变更用 Alembic（SQLAlchemy）或 Django migrations，不要手工改生产库。迁移脚本要能回滚，且在大表上加索引要考虑锁表时间（PostgreSQL 可用 CREATE INDEX CONCURRENTLY）。

7. 异步数据库访问

同步驱动会阻塞事件循环。异步应用要用异步驱动：

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/app")
    rows = await conn.fetch("SELECT id, name FROM users WHERE id = $1", 1)
    print(rows)
    await conn.close()

asyncio.run(main())

或 SQLAlchemy 异步引擎：

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/app")

# 片段：User 省略
async def main():
    async with AsyncSession(engine) as session:
        result = await session.execute(select(User))
        print(result.scalars().all())
    await engine.dispose()

asyncio.run(main())

注意：

• asyncpg 的占位符是 $1，不是 %s；
• 异步 session 不能在多个协程间共享；
• 事务边界要清晰，避免一个请求里开多个并行事务互相锁。

8. 应用开发中的组合

场景	做法
列表页分页	LIMIT/OFFSET 小数据量；深分页用「上一页最后一个 id」的 keyset 分页
计数	COUNT(*) 大表很贵；用近似统计或维护计数表
批量写入	executemany / COPY / INSERT ... ON CONFLICT，不要逐条 commit
幂等	唯一索引 + ON CONFLICT DO NOTHING
软删除	deleted_at IS NULL 放进联合索引
多租户	所有查询带 tenant_id，索引以 tenant_id 开头
缓存	先查 Redis，再查库；注意缓存穿透/雪崩

9. 常见坑

• f-string 拼 SQL：注入 + 执行计划无法复用。
• with conn 当成关闭连接：在 sqlite3 / psycopg2 里它只提交/回滚事务；SQLAlchemy 的 with engine.connect() / with Session(...) 语义不同。
• 长事务：外部 API、文件上传、睡眠都放在事务里。
• 连接泄漏：异常路径没有 close/rollback，池子逐渐耗尽。
• SELECT *：多读列、覆盖索引失效、schema 变更影响大。
• N+1：ORM 懒加载在循环里触发大量查询。
• 时区：数据库存 UTC，展示层转本地时区；不要混用 timestamp 和 timestamptz。
• 池子大于数据库容量：所有 worker 一起抢连接，数据库先崩。

10. 动手实验

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT, name TEXT)")
conn.executemany(
    "INSERT INTO users (email, name) VALUES (?, ?)",
    [(f"u{i}@example.com", f"user{i}") for i in range(10000)],
)
conn.commit()

# 没索引：SCAN
print("before index:")
for row in conn.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ("u500@example.com",)):
    print(row)

conn.execute("CREATE INDEX idx_users_email ON users(email)")

# 有索引：SEARCH ... USING INDEX
print("after index:")
for row in conn.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ("u500@example.com",)):
    print(row)

观察两件事：

• 执行计划从 SCAN users 变成 SEARCH users USING INDEX idx_users_email；
• 索引让读变快，但插入/更新要多维护一棵 B+ 树，写放大是代价。

11. 小结

• DB-API 是连接、游标、参数、事务的最小共同接口；不同驱动的占位符不同。
• 参数化查询是安全底线；表名/列名用白名单，不参与占位符。
• 事务边界要短；在 sqlite3 / psycopg2 里 with conn 管事务、不负责关闭连接，SQLAlchemy 的 with 语义要单独看。
• 索引是 B+ 树，不是 dict；用 EXPLAIN 验证，联合索引遵守最左前缀。
• 连接池是并发闸门，workers × (pool_size + max_overflow) 不能超过数据库容量。
• ORM 提升开发效率，但 N+1、懒加载、Session 生命周期要自己管。