PostgreSQL 与 SQLAlchemy 2
何时使用
- 设计表、约束和索引时。
- 编写 SQLAlchemy 2 模型和查询（同步或异步）时。
- 查询性能缓慢、锁堆积或连接耗尽时。
- 编写 Alembic 迁移，特别是针对有实时流量的表进行操作时。
对于破坏性数据变更，还应应用 naki-safety。如果项目有项目规则技能（例如 naki-myapp），请遵循它。
模式规则
- 主键：
bigint generated always as identity或 UUID（如果可用，使用 v7 以获得索引局部性）。如果枚举很重要，不要公开暴露顺序 ID。 - 类型：
timestamptz（永远不要使用无时区的timestamp），text配合CHECK长度约束代替varchar(n)（当限制可能改变时），货币使用numeric，jsonb（而不是json），boolean not null default false。 - 约束是最后一道防线： 默认使用
NOT NULL，UNIQUE，CHECK，外键需明确选择ON DELETE行为。 - 为每个用于连接或级联的外键创建索引；PostgreSQL 不会自动创建。
- 通过 SQLAlchemy
MetaData中的命名约定来命名约束和索引，以确保 Alembic 差异生成的稳定性。 jsonb用于真正可变的数据，而不是为了避免建模。查询频繁的键应成为真实列。- 只在需要时使用软删除； 如果使用，添加部分索引
WHERE deleted_at IS NULL。
索引规则
| 查询模式 | 索引 |
|---|---|
| 单列等值查询和范围查询 | B-tree |
WHERE a = ? AND b > ? ORDER BY b | 复合索引 (a, b)，等值列在前 |
| 行子集（例如开放项目） | 部分索引 WHERE status = 'open' |
lower(email) = ? | lower(email) 上的表达式索引 |
jsonb 包含关系 @> | GIN |
全文搜索，相似度 ILIKE '%x%' | 使用 tsvector 或 pg_trgm 的 GIN 索引 |
| 仅追加的大数据量时间序列 | 时间戳上的 BRIN 索引 |
- 每个索引都会增加写入开销和磁盘占用。移除未使用的索引（长时间窗口内
pg_stat_user_indexes.idx_scan = 0）。 - 使用
INCLUDE (...)的覆盖索引，实现热读取场景下的索引-only 扫描。
查询调优工作流
- 获取真实查询（启用
echo或log_min_duration_statement）和真实参数。 - 在生产级数据集上运行
EXPLAIN (ANALYZE, BUFFERS)。永远不要相信只有 100 行数据的开发表上的执行计划。 - 查找：对大表使用选择性过滤器的
Seq Scan，行估计与实际相差甚远（统计信息过时：需要ANALYZE），遍历多行的嵌套循环，排序导致磁盘溢出，高shared read。 - 修复成本最低的原因：添加或调整索引，将谓词重写为可索引形式（sargable），消除 N+1 问题，使用键集分页。
- 重新运行
EXPLAIN ANALYZE并比较时间。使用pg_stat_statements查找总时间最长的查询。
SQLAlchemy 2 规则
- 使用 2.0 风格：
select()，session.execute()，session.scalars()，Mapped[...]和mapped_column。 - 每个工作单元（请求、任务）使用一个会话。 禁止在多个线程或任务之间共享会话。
- 避免 N+1 问题： 使用
selectinload（集合）或joinedload（多对一）显式加载关系。在异步代码中，惰性加载会引发异常；在同步代码中，设置lazy="raise"可以捕获它。 - 仅使用绑定参数。永远不要将用户输入格式化为 SQL 字符串，包括
text()。 - 设置连接池大小和
pool_pre_ping=True；所有应用实例的总连接数必须保持在max_connections以下（如果工作进程很多，请使用 PgBouncer）。 - 保持事务简短；在持有事务期间不要调用其他服务的网络请求。
from datetime import datetime
from sqlalchemy import ForeignKey, Index, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload
class Base(DeclarativeBase):
pass
class Incident(Base):
__tablename__ = "incidents"
id: Mapped[int] = mapped_column(primary_key=True)
status: Mapped[str] = mapped_column(String(16), index=True)
opened_at: Mapped[datetime]
alerts: Mapped[list["Alert"]] = relationship(back_populates="incident", lazy="raise")
__table_args__ = (Index("ix_incidents_open", "opened_at", postgresql_where="status = 'open'"),)
async def open_incidents(session: AsyncSession, limit: int) -> list[Incident]:
stmt = (
select(Incident)
.where(Incident.status == "open")
.order_by(Incident.opened_at.desc(), Incident.id.desc())
.limit(limit)
.options(selectinload(Incident.alerts))
)
return list((await session.scalars(stmt)).all())
事务与锁
- 默认的
READ COMMITTED隔离级别适用于大多数场景。使用SELECT ... FOR UPDATE锁定将要根据读取内容进行修改的行。 - 对于队列：
FOR UPDATE SKIP LOCKED允许多个工作进程在不相互阻塞的情况下领取任务。 - Upsert 操作：使用
INSERT ... ON CONFLICT (key) DO UPDATE代替先读取后插入的竞争条件写法。 - 为应用程序角色设置
lock_timeout和statement_timeout，防止单个不良查询阻塞所有操作。 - 死锁通常由不一致的行锁定顺序引起；应使用一致的顺序（例如按 id 排序）来锁定行。
Alembic 与零停机变更
始终审查自动生成的迁移；自动生成会遗漏重命名操作（它会生成删除并添加）以及某些类型变更。
对于破坏性变更，采用扩展和收缩策略：
- 扩展： 添加新的可空列（或带有常量默认值，在 PG 11+ 上是即时操作），新表，或使用
CONCURRENTLY创建新索引。 - 部署同时读写新旧字段的代码。
- 回填： 在迁移之外分批处理（例如每事务 5k 行），或在单独的数据迁移中处理并记录进度。
- 部署仅读取新字段的代码。
- 收缩： 在后续版本中删除旧列。
对大型活跃表的高危操作：
| 操作 | 更安全的方法 |
|---|---|
CREATE INDEX | CREATE INDEX CONCURRENTLY（Alembic：在 with op.get_context().autocommit_block(): 内部设置 postgresql_concurrently=True） |
添加 NOT NULL | 添加 CHECK (col IS NOT NULL) NOT VALID，然后 VALIDATE CONSTRAINT，最后设置 NOT NULL |
| 添加外键 | 使用 NOT VALID 添加，然后 VALIDATE CONSTRAINT |
| 重命名列 | 采用扩展和收缩策略，绝不能在旧代码仍在运行时原地重命名 |
| 更改列类型 | 添加新列并进行回填，不要使用 ALTER TYPE 重写整个表 |
- 在迁移中设置
SET lock_timeout = '5s'，使它们在长时间事务后排队时快速失败。 - 每个迁移都必须有可用的
downgrade()或书面说明为什么无法降级。 - 在真实数据的副本上测试迁移的升级、降级和再次升级。
检查清单
- 使用
timestamptz，NOT NULL默认值，命名约束，外键已建索引。 - 查询已经参数化；SQL 中没有 f-string。
- 无 N+1 问题；关系已显式加载。
- 慢查询已使用
EXPLAIN (ANALYZE, BUFFERS)在真实数据上检查过。 - 事务简短；已设置超时。
- 迁移已审查、可逆，在需要的地方使用了
CONCURRENTLY/NOT VALID。 - 回填操作已分批并且可恢复。
常见错误
- 使用
OFFSET 100000进行分页；应在已索引的排序列上使用键集分页。 - 使用
WHERE date(created_at) = ?导致索引失效；应对原始列使用范围查询。 - 使用
LIKE '%term%'而未使用相似度索引。 - 忘记
await session.commit()，或在循环中每行提交一次。 - 运行
alembic revision --autogenerate并未经审查直接提交。 - 在调用外部 API 时保持事务打开。
- 使用本地时间存储时间戳。