sqlalchemy

sqlalchemy — SKILL.md

バリアント： standard · 使用するタイミング： スキルが呼び出され、動作するマルチダイアレクトの SQLAlchemy データレイヤー（またはその関連部分）を生成し、制御が呼び出し元に戻る場合です。

概要

このスキルでは、SQLAlchemy 2.x を使った移植性の高いリレーショナルデータレイヤーの構築方法を解説します。モデル、セッション、クエリを変更せずに SQLite、PostgreSQL、MySQL で実行できる単一のコードベースです。中心となるのは 型付きの 2.x ORM（DeclarativeBase、Mapped[...]、mapped_column(...)）で、ORM では移植性を保って表現できない少数の機能（ダイアレクト固有のアップサート、RETURNING）に限り、Core を使用します。基本は同期処理で、主な例も同期処理です。非同期処理については簡潔に補足します。重要な内容は ダイアレクト間の落とし穴をまとめた表です。行ロック、JSON、アップサート、ID、分離レベルなど、エンジンごとに挙動が異なるため、「一度書けば 3 種類のデータベースで動く」という前提が崩れる原因となる挙動を取り上げます。

使用するタイミング

• ✅ SQLite、PostgreSQL、MySQL のいずれかを対象とするアプリ向けに、SQLAlchemy の Engine、型付き DeclarativeBase／Mapped モデル、セッションスコープのトランザクションを構築する場合。
• ✅ ダイアレクトに適した DBAPI ドライバー（同期または非同期）を選び、接続 URL を記述する場合。
• ✅ 移植性の高いデータレイヤーのプーリング、トランザクションスコープ、または分離レベルを設定する場合。
• ✅ 「SQLite では動くが Postgres では失敗する」（またはその逆）という落とし穴（行ロック、JSON、アップサート、自動採番、分離レベル）に遭遇した場合。

次の場合は有効にしないでください：

• スキーマの マイグレーション の作成・実行や ALTER TABLE による変更 → alembic。
• ジョブキュー／スケジューラーの構築（対象ジョブのクエリ、アトミックなリース処理ループ、ハートビート、公平なリソース割り当て）→ sql-job-queue。（このスキルで解説するのは with_for_update(skip_locked=...) の プリミティブ のみです。）
• Pydantic によるデータの検証／シリアライズ → pydantic-v2。
• 別の ORM（Django ORM、Tortoise、SQLModel）または生の DBAPI を使用する場合。

ワークフロー

ステップ 1：ダイアレクトとドライバーを選び、URL を記述する

URL の形式は dialect+driver://user:password@host:port/dbname です。ダイアレクトごとの デフォルト ドライバーを選びます（ステップ 5 の表を参照）。ドライバーは +driver の部分です。

# SQLite — file (the +pysqlite / sqlite3 stdlib driver is the default)
"sqlite+pysqlite:///./app.db"          # relative file
"sqlite:///./app.db"                    # same thing — sqlite3 is the default driver
"sqlite:////absolute/path/app.db"       # note the 4th slash = absolute path
"sqlite://"                             # in-memory (also sqlite:///:memory:)

# PostgreSQL — psycopg (v3, the modern maintained driver)
"postgresql+psycopg://user:pw@localhost:5432/appdb"

# MySQL / MariaDB — PyMySQL (pure-Python, no C build step)
"mysql+pymysql://user:pw@localhost:3306/appdb"

ステップ 2：プロセスごとにエンジンを 1 つ作成する

Engine は接続プールを管理します。プロセスの開始時に一度だけ作成し、すべての場所で共有してください。リクエストやタスクごとにエンジンを作成しないでください。

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg://user:pw@localhost/appdb",
    pool_pre_ping=True,      # recycle dead connections (server DBs — PG/MySQL)
    pool_size=5,             # persistent pooled connections
    max_overflow=10,         # extra burst connections beyond pool_size
)

プーリングのデフォルト設定はダイアレクトによって異なります（「注意点」を参照）。サーバーデータベース（PG／MySQL）では pool_pre_ping=True を設定してください。サーバーのアイドルタイムアウトで切断された接続を検出して置き換え、クエリの実行中にエラーが発生するのを防げます。

ステップ 3：型付きの 2.x ORM モデルを定義する

DeclarativeBase を継承し、列に Mapped[...] の型注釈を付けて mapped_column(...) で設定します。従来の Column = ...／declarative_base() 形式は 絶対に使用しないでください。

from datetime import datetime
from sqlalchemy import JSON, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[int] = mapped_column(primary_key=True)          # autoincrements on all 3
    name: Mapped[str] = mapped_column(index=True)              # NOT NULL (no Optional)
    data: Mapped[dict] = mapped_column(JSON)                   # portable JSON (Step 7)
    note: Mapped[str | None]                                   # NULLable (Optional -> nullable)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    runs: Mapped[list["Run"]] = relationship(back_populates="job")

class Run(Base):
    __tablename__ = "runs"
    id: Mapped[int] = mapped_column(primary_key=True)
    job_id: Mapped[int] = mapped_column(ForeignKey("jobs.id"))
    job: Mapped[Job] = relationship(back_populates="runs")

Mapped[str] は NOT NULL、Mapped[str | None] は NULL 許容です。SQLAlchemy は型注釈から NULL 許容性を判断します。整数の主キーは、すべてのダイアレクトで自動採番されます（ステップ 9）。

ステップ 4：スキーマを初期化する（開発／テストのみ）。その後、セッションとクエリを設定する

Base.metadata.create_all(engine)   # DEV/TEST bootstrap ONLY — see Anti-patterns

実際のプロジェクトで管理対象のスキーマを変更する場合は、alembic を使用してください。マイグレーション手段として create_all() をリリースしないでください。

作業単位ごとに短時間で終了するセッションを使用し、トランザクションを明示します。

from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

SessionLocal = sessionmaker(engine)   # a factory; bind the engine once

# Per unit of work: open, transact, close.
with Session(engine) as session, session.begin():   # begin() = commit on exit, rollback on error
    session.add(Job(name="ingest", data={"src": "s3"}))
    # block exit commits automatically

# Reads
with Session(engine) as session:
    job = session.get(Job, 1)                                    # typed Job | None by PK
    rows = session.execute(
        select(Job).where(Job.name == "ingest").order_by(Job.id)
    ).scalars().all()                                            # list[Job]

session.execute(select(...)) は Row タプルを返します。.scalars() を呼び出すと、ORM エンティティを取り出せます。session.get(Model, pk) は主キーによる検索の高速な方法で、まず ID マップを確認します。

ステップ 5：同期か非同期か、ドライバーを選ぶ（一覧表）

基本は同期です。 周囲のスタックがすでに非同期の場合（ASGI アプリなど）に限り、非同期を選択してください。URL の +driver で同期か非同期かが決まります。フラグはありません。

方言	デフォルトの同期ドライバー	URL プレフィックス	デフォルトの非同期ドライバー	非同期 URL プレフィックス	代替案（簡単な理由）
SQLite	sqlite3（pysqlite、標準ライブラリ）	sqlite+pysqlite:// / sqlite://	aiosqlite	sqlite+aiosqlite://	—（標準ライブラリのドライバーが、主流の同期ドライバーとして唯一の選択肢です）
PostgreSQL	psycopg（v3）	postgresql+psycopg://	asyncpg	postgresql+asyncpg://	psycopg2（レガシーですが成熟しており、同期専用です。既存の v2 デプロイでのみ使用してください）；psycopg v3 も非同期接続に使えるため、両方に同じドライバーを使いたい場合に適しています
MySQL / MariaDB	PyMySQL	mysql+pymysql://	aiomysql	mysql+aiomysql://	mysqlclient（C 拡張で高速ですが、ビルドツールチェーンが必要です）；asyncmy（非同期処理がより高速ですが、Cython のビルドが必要です）

デフォルトは、メンテナンスが継続され、導入時の手間が最小限であることを基準に選んでいます。psycopg v3 は現在も積極的にメンテナンスされている最新の PG ドライバーです。PyMySQL は純粋な Python 製のため、コンパイラーなしでインストールできます。asyncpg は最速の非同期 PG ドライバーです。

ステップ 6：非同期について（簡単に）

モデルクラスは同じまま、エンジンとセッションのファミリーを切り替え、I/O を await します。

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

aengine = create_async_engine("postgresql+asyncpg://user:pw@localhost/appdb")
ASessionLocal = async_sessionmaker(aengine, expire_on_commit=False)

async with ASessionLocal() as session:
    async with session.begin():
        session.add(Job(name="async-ingest", data={}))
    res = await session.execute(select(Job))
    jobs = res.scalars().all()

非同期処理では、greenlet ブリッジを介して同期 DBAPI の処理を実行します。await を使えば、SQLAlchemy が切り替えを処理します。注意点は、コミット後の属性アクセスによって遅延読み込みが発生する場合があり、非同期では（暗黙的な I/O が許可されないため）例外になることです。expire_on_commit=False を設定するか、コミット後にアクセスするリレーションシップを先行読み込み（selectinload）してください。1 つの AsyncSession を複数の並行タスクで共有しないでください（同期セッションを複数のスレッドで共有しないのと同じルールです）。

ステップ 7：方言ごとの注意点 — 違いがある部分は分岐する

ここは重要なセクションです。それぞれの詳しい例については、references/sqlalchemy-extras.md を参照してください。移植可能なルールは次のとおりです。

行ロック — 対応状況（重要）。 select(...).with_for_update(...) は SELECT ... FOR UPDATE を生成します。

stmt = select(Job).where(Job.name == "ingest").with_for_update(
    skip_locked=True,   # skip rows another txn holds (claim-next pattern)
    nowait=False,       # True = error instead of waiting if a row is locked
    read=False,         # True = FOR SHARE instead of FOR UPDATE
    of=Job,             # lock only this table's rows in a multi-table select
)

方言	FOR UPDATE	SKIP LOCKED / NOWAIT
PostgreSQL	対応	対応
MySQL	対応（InnoDB）	対応 — MySQL 8.0+ のみ
SQLite	非対応 — 行レベルのロックはありません（ファイル／データベースレベルのロックです）。with_for_update は実質的に何もしません。

SQLite でも動作させる必要があるデータレイヤーでは、行ロックに依存できません。SQLite で代わりに行う方法（BEGIN IMMEDIATE による確保戦略）は、兄弟プロジェクトの sql-job-queue の担当範囲です。ここでは分岐してそちらを参照し、実装しないでください。

JSON。 移植性のため、汎用の sqlalchemy.JSON を使います。PostgreSQL／MySQL ではネイティブ JSON として、SQLite では TEXT（SQLite の JSON1 経由）として保存されます。移植可能な範囲は、ドキュメント全体の保存と取得です。PostgreSQL ネイティブのインデックス／包含演算を使う場合は、代わりに sqlalchemy.dialects.postgresql.JSONB を使用しますが、これは PostgreSQL 専用です。パス／包含クエリの演算子は方言ごとに異なります。一つのエンジン向けに書いた JSON の WHERE 句が、別のエンジンでも動くとは限りません。

Upsert — 共通の書式はありません。方言ごとに分岐します。 方言モジュールから insert をインポートします。

# PostgreSQL (and SQLite — same API, different import)
from sqlalchemy.dialects.postgresql import insert   # or .sqlite import insert
stmt = insert(Job).values(id=1, name="ingest", data={})
stmt = stmt.on_conflict_do_update(index_elements=["id"], set_={"data": stmt.excluded.data})
# stmt.on_conflict_do_nothing(index_elements=["id"])   # the do-nothing variant
# index_elements must name a real UNIQUE / PK constraint (or partial-index cols), else no conflict is detected

# MySQL
from sqlalchemy.dialects.mysql import insert
stmt = insert(Job).values(id=1, name="ingest", data={})
stmt = stmt.on_duplicate_key_update(data=stmt.inserted.data)

「挿入されるはずだった行」を表すために、excluded（PostgreSQL／SQLite）と inserted（MySQL）を使い分ける点に注意してください。

自動インクリメント／ID。 整数の主キー（Mapped[int] = mapped_column(primary_key=True)）は、3 つすべてで自動インクリメントされます。PostgreSQL では IDENTITY／SERIAL、MySQL では AUTO_INCREMENT、SQLite では ROWID を使用します。移植可能なデフォルト設定でそのまま動作します。シーケンス処理を独自に実装しないでください。

トランザクション分離レベル。 エンジンごと、または接続ごとに設定します。

engine = create_engine(url, isolation_level="REPEATABLE READ")
# or per-connection / per-block:
with engine.connect().execution_options(isolation_level="SERIALIZABLE") as conn:
    ...

有効なレベルは READ UNCOMMITTED、READ COMMITTED、REPEATABLE READ、SERIALIZABLE、AUTOCOMMIT です。デフォルトは方言ごとに異なります。PostgreSQL は READ COMMITTED、MySQL/InnoDB は REPEATABLE READ です。SQLite では、設定可能な isolation_level として SQLAlchemy が標準でサポートするのは AUTOCOMMIT のみです（それ以外では、エンジンはおおむねシリアライザブルです）。isolation_level="SERIALIZABLE" が機能するとは考えないでください。正しさが特定のレベルに依存する場合は、明示的に設定し、方言のデフォルトを引き継がないでください。

ルール

厳守するルール（違反禁止）：

• 2.x の型付き ORM のみを使用します。 DeclarativeBase + Mapped[...] + mapped_column(...) を使います。従来の Column = ／ declarative_base() 形式は使用しないでください。
• プロセスごとにエンジンを 1 つだけ使用します。 エンジンが接続プールを所有します。エンジンは一度だけ作成し、全体で共有してください。
• セッションは短時間だけ使用し、共有しません。 作業単位ごとに Session／AsyncSession を 1 つ使用します。スレッド間や非同期タスク間でセッションを共有しないでください。
• create_all() は開発／テスト用の初期化にのみ使用します。 管理されたスキーマ変更には alembic を使用します。
• 方言ごとの注意点に応じて分岐します。 行ロック、単一の upsert 書式、JSON クエリ演算子、分離レベルのデフォルトが SQLite／PostgreSQL／MySQL 間で共通だと決めつけないでください。

推奨事項（必要に応じて変更可能）：

• 明示された制約（既存の psycopg2 デプロイ、パフォーマンスが重要な非同期処理など）が別の選択を正当化しない限り、方言ごとのデフォルトドライバー（ステップ 5）を優先してください。
• ORM を優先し、ORM では移植可能な形で表現できないもの（方言固有の upsert、RETURNING）に限り Core を使用してください。
• PostgreSQL／MySQL では pool_pre_ping=True を使用してください。SQLite では不要です。
• 同期処理を基本とし、すでに非同期スタックである場合に限り、非同期処理を使用してください。

注意事項

• sqlite:///path のスラッシュ数。 スラッシュ3つ＋相対パスは CWD からの相対パスです。絶対パスにはスラッシュが4つ必要です（sqlite:////abs/path.db）。インメモリの場合は sqlite:// または sqlite:///:memory: を使います。
• 新しい sqlite:// は接続ごとに空です。 通常のインメモリエンジンでは、プール内の各接続がそれぞれ独自のデータベースを持ちます。ある接続で作成したテーブルは、次の接続からは見えません。共有インメモリ DB（テスト用）には、StaticPool（共有接続1つ）を使います：create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)。
• プールのデフォルトは方言によって異なります。 ファイル SQLite → QueuePool（2.0 以降のデフォルト）、:memory: → SingletonThreadPool、共有インメモリ／テスト → StaticPool。サーバー DB では QueuePool が使われます。どこでも同じプールだと思い込まないでください。
• SQLite では with_for_update は何もせずに終了します。 エラーにはならず、ロックも行いません。SQLite で「動作する」コードでも競合状態が起きる可能性があるため、実際に使うエンジンで確認してください。
• コミット後の非同期遅延読み込みは例外を発生させます。 非同期コミットの後に未ロードのリレーション／属性へアクセスすると、非同期処理では禁止されている暗黙の I/O が発生します。expire_on_commit=False を設定するか、あらかじめ Eager Load してください。
• .execute(select(Model)) はエンティティではなく行を返します。 .scalars() を忘れると、Job ではなく1要素タプル（(Job,)）が返されます。
• MySQL の SKIP LOCKED には 8.0 以降が必要です。 5.7 では何もせずに終了するのではなく、構文エラーになります。

アンチパターン

• 「より簡単だから」と create_all() をマイグレーション手段としてリリースしないでください。スキーマを進化させることはできません。alembic を使ってください。
• すべての方言で共通の API が使えるかのような「移植可能な upsert」ヘルパーを書かないでください。excluded／ON CONFLICT（PG／SQLite）と inserted／ON DUPLICATE KEY UPDATE（MySQL）は実際に異なります。方言に応じて分岐してください。
• SQLite が行をロックすると想定して、with_for_update だけで次のジョブを取得するループを組まないでください。行はロックされません。それは sql-job-queue の役割です。
• チュートリアルやソーススキルで見たからといって、レガシーな Column =／declarative_base() スタイルを使わないでください。これは 2.0 以前の形式で、型付けの利点が失われます。
• 「オーバーヘッドを避けるため」に、アプリ全体で長時間存続する Session を1つだけ開かないでください。セッションのコストは低く、長時間存続するセッションは古いアイデンティティマップの状態を残し、並行処理を妨げます。
• リクエスト／タスクごとにエンジンを作成しないでください。コネクションプールが使われず、コネクションが枯渇します。

出力

複数方言に対応した、動作する SQLAlchemy 2.x データレイヤー（または要求された範囲）。プロセスごとに、適切な URL とプール設定を備えた Engine／AsyncEngine、型付けされた DeclarativeBase／Mapped モデル、短時間のセッションスコープのトランザクション、理由を明示して選んだ同期または非同期ドライバー、そして方言間の各種注意事項に対応する方言ごとの分岐を含みます。この成果物は次のレイヤーで使われます。スキーマの進化にはマイグレーションツール（alembic）、またはこれらの基本要素を組み合わせる、より高レベルのパターン（例：sql-job-queue）です。

関連項目

• alembic — スキーママイグレーション／管理対象の ALTER 変更（SQLite で ALTER が使えず、バッチで移動・コピーする際の注意点を含む）。このスキルでは開発／テストの初期化に create_all() を使うところまでを扱い、管理対象の変更はそちらに委ねます。
• sql-job-queue — ジョブストア／スケジューラーのパターン：準備済みセットのクエリ、アトミックなリースのループ、ハートビート／クラッシュ後の再開、公平なリソース配分、そして with_for_update でロックできない SQLite での対処方法。このスキルでは with_for_update(skip_locked=...) の基本機能と、対応状況の一覧のみを扱います。
• pydantic-v2 — アプリケーション境界でのデータ検証／シリアライズ（ORM のカラム型ではありません）。

段階的な開示

• references/sqlalchemy-extras.md — 方言間の注意事項を網羅した、より詳しい実例（方言ごとの upsert の一連の例、JSON／JSONB、分離レベル、StaticPool のテスト設定、RETURNING のための Core への切り替え）。特定の注意事項を実装する際に、本文中のスニペットだけでは不十分な場合に読み込んでください。
• references/sources.md — このスキルの調査情報の出典。主張の出典を確認する場合や、新たにレビューする場合に読み込んでください。

本文の上限

• description は 1,024 文字以下。
• 本文は約 500 行／5,000 トークン以下。
• 参照ファイルごとに、10k トークンを超える場合は警告、25k を超える場合はエラー。参照ファイルの合計は、25k トークンを超える場合は警告、50k を超える場合はエラー。