PostgreSQL 與 SQLAlchemy 2
使用時機
- 設計資料表、限制條件與索引。
- 撰寫 SQLAlchemy 2 模型與查詢（同步或非同步）。
- 查詢緩慢、鎖定堆積或連線耗盡。
- 撰寫 Alembic 遷移，特別是針對有即時流量的資料表。
對於破壞性資料變更，也請套用 naki-safety。如果專案有專案規則技能（例如 naki-myapp），請遵循該技能。
Schema 規則
- 主鍵：
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選項的外部鍵。 - 為每個用於 JOIN 或串聯的外部鍵建立索引； PostgreSQL 不會自動建立。
- 透過 SQLAlchemy
MetaData的命名慣例來命名限制條件與索引， 以確保 Alembic Diff 穩定。 - 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 (...)的覆蓋索引以進行僅索引掃描。
查詢調校工作流程
- 取得真實查詢（啟用
echo或log_min_duration_statement）與真實參數。 - 在類似生產環境的資料集上執行
EXPLAIN (ANALYZE, BUFFERS)。絕不相信來自僅有 100 筆資料的開發資料表的執行計畫。 - 尋找：大型資料表上帶有選擇性過濾條件的
Seq Scan、與實際資料筆數相差甚遠的預估筆數（統計資料過時：執行ANALYZE）、多筆資料的巢狀迴圈、排序溢寫至磁碟、高shared read。 - 修復最便宜的成因：新增或調整索引、改寫謂詞使其可搜尋、消除 N+1、使用 keyset 分頁。
- 重新執行
EXPLAIN ANALYZE並比較時間。使用pg_stat_statements尋找總時間最長的查詢。
SQLAlchemy 2 規則
- 使用 2.0 風格：
select()、session.execute()、session.scalars()、Mapped[...]與mapped_column。 - 每個工作單元（請求、任務）使用一個 Session。 絕不在執行緒或任務之間共享 Session。
- 避免 N+1： 明確使用
selectinload（集合）或joinedload（多對一）載入關聯。在非同步程式中，惰性載入會引發例外；在同步程式中設定lazy="raise"以捕捉此問題。 - 僅使用繫結參數。絕不將使用者輸入格式化為 SQL 字串，包括
text()。 - 設定連線池大小與
pool_pre_ping=True；所有應用程式實例的總連線數必須維持低於max_connections（如果 Worker 很多，請使用 PgBouncer）。 - 保持交易簡短；在持有交易時，不得呼叫其他服務的網路 API。
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允許多個 Worker 領取工作而不互相封鎖。 - Upsert：使用
INSERT ... ON CONFLICT (key) DO UPDATE而不是先讀後插的競爭條件。 - 為應用程式角色設定
lock_timeout與statement_timeout，以免一個錯誤查詢癱瘓所有功能。 - 死結來自於以不同順序鎖定資料列；請以一致的順序（例如依 ID 排序）鎖定。
Alembic 與零停機變更
務必審查自動產生的遷移；自動產生會遺漏重新命名（它會發出刪除再加新增）以及某些型別變更。
對於破壞性變更，請使用擴展與收縮模式：
- 擴展： 新增可為空的資料行（或帶有常數預設值，這在 PG 11+ 上是即時的）、新資料表或使用
CONCURRENTLY的新索引。 - 部署同時寫入新舊兩者的程式碼。
- 回填 以批次方式（例如每次交易 5000 筆）在遷移外部進行，或在包含進度記錄的獨立資料遷移中進行。
- 部署讀取新資料的程式碼。
- 收縮： 在後續的版本中刪除舊資料行。
在大型即時資料表上的危險操作：
| 操作 | 更安全的方法 |
|---|---|
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分頁；請在已建立索引的排序鍵上使用 keyset 分頁。 - 使用
WHERE date(created_at) = ?導致索引失效；請在原始資料行上使用範圍查詢。 - 使用
LIKE '%term%'但未建立三元組索引。 - 忘記執行
await session.commit()，或在迴圈中逐筆提交。 - 執行
alembic revision --autogenerate並在未經審查的情況下提交。 - 在呼叫外部 API 時保持交易開啟。
- 以本地時間儲存時間戳記。