PostgreSQL と SQLAlchemy 2
使用するタイミング
- テーブル、制約、インデックスを設計するとき。
- SQLAlchemy 2 のモデルとクエリを書くとき（同期・非同期両方）。
- クエリが遅い、ロックがたまる、または接続が枯渇するとき。
- Alembic マイグレーションを書くとき、特に稼働中のトラフィックが発生しているテーブルに対して。
破壊的なデータ変更には、naki-safety も適用してください。プロジェクトにプロジェクトルールスキルがある場合（例：naki-myapp）はそれに従ってください。
スキーマルール
- 主キー：
bigint generated always as identityまたは UUID（可能なら v7、インデックスの局所性のため）。列挙が問題になる場合はシーケンシャルIDを公開しないでください。 - 型：
timestamptz（タイムゾーンなしのtimestampは使わない）、長さ制限が変わる可能性がある場合はvarchar(n)の代わりにtextと長さのCHECK、金額にはnumeric、jsonb（jsonではない）、boolean not null default false。 - 制約は最後の防御線： デフォルトで
NOT NULL、UNIQUE、CHECK、明示的なON DELETEを指定した外部キー。 - 結合やカスケードで使われる外部キーにはすべてインデックスを張る。PostgreSQL は自動的には行いません。
- 制約とインデックスに命名規則を適用する。SQLAlchemy の
MetaDataで命名規則を設定すると、Alembic の差分が安定します。 - 本当に可変なデータには
jsonbを使う。モデリングを避けるためではありません。クエリがホットなキーは実際のカラムにします。 - 本当に必要な場合のみソフトデリートを使う。使う場合は、部分インデックス
WHERE deleted_at IS NULLを追加する。
インデックスルール
| クエリパターン | インデックス |
|---|---|
| 1カラムに対する等価・範囲 | B-tree |
WHERE a = ? AND b > ? ORDER BY b | 複合 (a, b)、等価カラムを先に |
| 行のサブセット（未処理項目） | 部分インデックス WHERE status = 'open' |
lower(email) = ? | 式インデックス lower(email) に対して |
jsonb 包含 @> | GIN |
全文検索、トライグラム ILIKE '%x%' | GIN（tsvector または pg_trgm を使用） |
| 巨大な追記専用時系列 | タイムスタンプカラムに BRIN |
- すべてのインデックスは書き込みとディスクを消費します。使われていないインデックスは削除してください（
pg_stat_user_indexes.idx_scan = 0を長期間確認）。 - ホットリードにおけるインデックスオンリースキャンのために
INCLUDE (...)を使用したカバリングインデックス。
クエリチューニングのワークフロー
- 実際のクエリ（
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。 - 作業単位（リクエスト、ジョブ）ごとに1つのセッション。セッションをスレッドやタスク間で共有しないでください。
- 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を設定し、1つの悪いクエリがすべてを停止させないようにします。 - デッドロックは行を異なる順序でロックすることから発生します。一貫した順序でロックしてください（IDでソート）。
Alembic とダウンタイムなしの変更
自動生成されたマイグレーションは常にレビューしてください。自動生成ではリネーム（drop + add を出力します）や一部の型変更を見逃します。
破壊的な変更には Expand and Contract を使用します。
- Expand： 新しいカラムを nullable（または定数デフォルト付き。PG 11+ では即時）で追加、新しいテーブル、または新しいインデックスを
CONCURRENTLYで追加。 - 新旧両方を書き込むコードをデプロイする。
- バックフィル をバッチ（例：トランザクションあたり 5000 行）で、マイグレーションの外か、進捗ログ付きの別のデータマイグレーションで行う。
- 新しい方を読み取るコードをデプロイする。
- Contract： 後のリリースで古いカラムを削除する。
大規模な稼働テーブルに対する危険な操作：
| 操作 | より安全なアプローチ |
|---|---|
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 |
| カラムのリネーム | Expand and Contract、古いコードが実行されている状態でのインプレース変更は絶対に行わない |
| カラム型の変更 | 新しいカラムとバックフィル。テーブルを書き換える ALTER TYPE は避ける |
- マイグレーションでは
SET lock_timeout = '5s'を設定し、長時間のトランザクションの後ろでキューイングされるのではなく、すぐに失敗するようにする。 - すべてのマイグレーションは動作する
downgrade()を持つか、それが不可能である理由を明記したメモを持つ。 - 現実的なデータのコピーに対して、マイグレーションの up、down、もう一度 up をテストする。
チェックリスト
-
timestamptz、NOT NULLデフォルト、名前付き制約、インデックス付き外部キー。 - クエリはパラメータ化されている。SQL 内に f-string はない。
- N+1 がない。リレーションシップは明示的にロードされている。
- 遅いクエリは、現実的なデータで
EXPLAIN (ANALYZE, BUFFERS)を使用して確認されている。 - トランザクションは短い。タイムアウトが設定されている。
- マイグレーションはレビューされ、ロールバック可能で、必要に応じて
CONCURRENTLY/NOT VALIDを使用している。 - バックフィルはバッチ化され、再開可能である。
よくある間違い
OFFSET 100000ページネーション。インデックス付きソートキーでキーセットページネーションを使用する。WHERE date(created_at) = ?でインデックスが無効になる。raw カラムに対して範囲を使用する。- トライグラムインデックスなしで
LIKE '%term%'を使用する。 await session.commit()を忘れる、またはループ内で行ごとにコミットする。alembic revision --autogenerateを実行し、読まずにコミットする。- 外部APIを呼び出している間、トランザクションを開いたままにする。
- タイムスタンプをローカル時間で保存する。