SQLAlchemy 2.0 実践ガイド：型安全ORMで本番運用に耐えるデータアクセス層を設計する

導入：なぜ今「SQLAlchemy 2.0」を学び直すべきなのか

SQLAlchemy は Python における事実上の標準 ORM です。しかし、Web 上に流通する解説記事・Stack Overflow の回答・生成 AI が出力するコードの多くは、いまだに 1.x 系のレガシースタイル（Column()、session.query()、Base = declarative_base()）で書かれています。

2023 年 1 月に正式リリースされた SQLAlchemy 2.0 は、単なるバージョンアップではなく、型システムを中核に据えたパラダイムシフトでした。Mapped[...] アノテーションによる完全な型安全、Core と ORM を統一した select() API、ネイティブな asyncio サポート——これらは「動けばいい」コードと「本番運用に耐える」コードを明確に分かつ分水嶺です。

この記事は、入門の繰り返しではありません。公式ドキュメント（docs.sqlalchemy.org/en/20）に忠実でありながら、それより一段わかりやすく、かつ実務で必ず直面する以下の壁を、具体的なコードで突破することを目的とします。

• 「session.query() で書いてきたが、2.0 スタイルの select() への移行で何が変わるのか分からない」
• 「Mapped[int] と mapped_column() の使い分け、Optional の扱いが曖昧」
• 「一覧画面を開くと SQL が大量に飛ぶ（N+1 問題）。selectinload と joinedload のどちらを使うべきか」
• 「FastAPI で async 化したらリレーションへのアクセスで MissingGreenlet エラーが出た」
• 「本番（特に RDS / サーバーレス）で connection closed 系のエラーが散発する」

筆者は、経済産業大臣賞を受賞した B2B SaaS のバックエンドを Python 3.11 / SQLAlchemy 2.0 / PostgreSQL 16 で設計・実装し、Router → UseCase → Repository → Model の厳格な層分離で本番運用してきました。本記事は、その実装で得た知見を公式ドキュメントの裏付けと共に整理したものです。

1. 型安全なモデル定義：DeclarativeBase と Mapped の正しい使い方

2.0 スタイルのモデル定義は、Python の型アノテーションと完全に統合されています。まず、公式の「ORM Quickstart」に忠実なモデルを示します。

from typing import List, Optional

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
)


class Base(DeclarativeBase):
    """全モデルの基底クラス。2.0では declarative_base() ではなく
    DeclarativeBase を継承するのが正攻法。"""


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]  # NULL許容（str | None でも可）

    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan",
    )

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r})"


class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]  # 型アノテーションだけで NOT NULL が決まる
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

    user: Mapped["User"] = relationship(back_populates="addresses")

H3: Mapped[...] が「型」と「NULL 制約」を同時に決める

ここが 1.x からの最大の変化点です。2.0 では、型アノテーション自体がスキーマ情報を持ちます。

email_address: Mapped[str] のように mapped_column() を省略しても、アノテーション str から String 型・NOT NULL 制約が自動導出されます。型情報を追加で制御したいとき（長さ・主キー・外部キーなど）にだけ mapped_column() を添えます。

なぜこれが優れているのか？ 1.x の name = Column(String(30)) は、IDE / 型チェッカー（mypy・Pyright）から見ると単なる Column インスタンスであり、user.name が str であることを静的に保証できませんでした。2.0 の Mapped[str] は、mypy が user.name を真に str として推論します。user.fullname（Optional[str]）に対して .upper() を呼べば型チェックで弾かれる——これは CLAUDE.md でいう「型安全（Type Safety）」を ORM 境界まで貫徹できることを意味します。

⚠️ レガシーとの違い：Column() は 2.0 でも動作しますが、Mapped[...] アノテーション下での正準は mapped_column() です。新規コードで Column() を混在させると型推論が効かず、技術的負債になります。

H3: type_annotation_map でプロジェクト規約を一元化する（DRY）

「int は常に BIGINT」「datetime は常にタイムゾーン付き」といったプロジェクト規約は、列ごとに書くと DRY 違反になります。DeclarativeBase の type_annotation_map で一箇所に集約できます。

import datetime
from sqlalchemy import BigInteger
from sqlalchemy.dialects.postgresql import TIMESTAMP


class Base(DeclarativeBase):
    type_annotation_map = {
        int: BigInteger,
        datetime.datetime: TIMESTAMP(timezone=True),
    }

これ以降、created_at: Mapped[datetime.datetime] と書くだけで、全モデルで自動的に TIMESTAMP WITH TIME ZONE が使われます。規約変更が一点で済む——典型的な「ETC（Easy To Change）」の実践です。

2. クエリの統一 API：session.query() を捨て、select() に移行する

2.0 の心臓部は、Core と ORM で同一の select() 構文を使えるようになったことです。

from sqlalchemy import select

# 2.0スタイル：エンティティを直接イテレートできる
stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

for user in session.scalars(stmt):
    print(user)

H3: session.scalars() と session.execute().scalars() の使い分け

これは公式ドキュメントでも混乱しやすい点なので、明確に整理します。

select(User) の結果を session.execute() で受け取ると、各行は 1 要素のタプル（Row） にラップされます。

session.execute(select(User)).all()
# → [(User(id=1),), (User(id=2),)]   ← タプルに包まれている

エンティティを「裸」で受け取りたい——これが大半のケースです。そのためのショートカットが session.scalars() です。公式の定義どおり、session.scalars(stmt) は session.execute(stmt).scalars() と完全に等価です。

session.scalars(select(User)).all()
# → [User(id=1), User(id=2)]         ← 裸のエンティティ

実務での指針：

select(User.name, User.fullname) のように複数カラムを取る場合は execute() を使い、Row タプルとして受け取ります。「単一エンティティなら scalars、複数カラムなら execute」と覚えるのが最短です。

H3: session.query() は「動くが、レガシー」

公式ドキュメントは Query オブジェクトを明確に legacy construct（as of SQLAlchemy 2.0） と位置づけています。重要なのは次のニュアンスです。

Query API の大部分は SQLAlchemy から削除されない。Query は実行時に内部で 2.0 スタイルの select() へ変換される、ごく薄いアダプタになった。

つまり既存コードが壊れることはありません。しかし新規コードを query() で書く理由はもはやなく、select() に統一することで Core クエリとの記述差がなくなり、学習コストと認知負荷（Cognitive Load）が下がります。

3. Session とユニットオブワーク：トランザクション境界を設計する

Session は SQLAlchemy ORM における「作業単位（Unit of Work）」の管理者です。本番コードでの正しいライフサイクル管理が、データ整合性の生命線になります。

H3: sessionmaker とコンテキストマネージャ

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/mydb")

# アプリ起動時に一度だけ生成するファクトリ
SessionFactory = sessionmaker(engine)

# リクエスト単位でセッションを開く（with で確実にクローズ）
with SessionFactory() as session:
    session.add(some_object)
    session.commit()

トランザクション境界を明示したい場合、session.begin() を併用します。begin() ブロックを抜けると commit、例外時は rollback が自動で行われます。

# 推奨：with の二段構えでトランザクションを明示
with SessionFactory() as session, session.begin():
    session.add(user)
    session.add(address)
    # ブロックを抜けると自動 commit、例外なら自動 rollback

なぜこれが優れているのか？ try/except/finally で手動 commit() / rollback() / close() を書くと、漏れが必ず発生します。コンテキストマネージャに委譲することで、「例外時は必ずロールバックされ、接続は必ず返却される」という回復性（Reliability）をコードレベルで保証できます。

H3: commit / flush / expire の罠を理解する

本番で最も多いバグの温床がここです。公式ドキュメントの挙動を正確に押さえます。

• flush：保留中の変更を SQL として DB に送る（が、commit はしない）。デフォルトではクエリ発行前に自動で flush（autoflush） される。
• commit：commit() は内部で無条件に flush() を呼んでから COMMIT を発行する。
• commit 後の expire：commit 後、セッション内の全オブジェクトは**期限切れ（expired）**になり、次回アクセス時に再ロードされる。

この「commit 後 expire」が罠です。次のコードを見てください。

with SessionFactory() as session, session.begin():
    user = User(name="alice")
    session.add(user)
# ブロックを抜けて commit 済み。session も閉じている。

print(user.name)  # ❌ DetachedInstanceError の危険

commit 後、user は expired 状態になり、属性アクセスで再ロードを試みます。しかし with を抜けてセッションが閉じているため、DetachedInstanceError が発生します。

対策は用途で分かれます。

# 対策1：commit 後も属性を使うなら expire_on_commit=False
SessionFactory = sessionmaker(engine, expire_on_commit=False)

# 対策2：必要な値を commit 前に取り出しておく（DTO へ詰め替える）

後述する async では、この「commit 後の暗黙的な再ロード」がさらに深刻な問題を引き起こすため、expire_on_commit=False がほぼ必須になります。

4. N+1 問題を根絶する：ローダー戦略の正しい選択

ORM を本番投入したときに最初に必ず刺さる性能問題が N+1 です。公式ドキュメントもこれを名指しで警告しています。

lazyload() 戦略は、ORM における最も一般的な問題の 1 つ、すなわち N+1 問題 を引き起こす。

H3: N+1 問題とは何か

リレーションのデフォルトは 遅延ロード（lazy="select"） です。次のコードは、一見無害に見えます。

users = session.scalars(select(User)).all()  # ① SELECT * FROM user_account
for user in users:
    print(user.addresses)  # ② user ごとに SELECT * FROM address WHERE user_id = ?

ユーザーが 100 人いれば、① の 1 回に加えて ② が 100 回——合計 101 回の SQL が飛びます。これが N+1 です。一覧画面のレスポンスが指数的に悪化する典型的なアンチパターンです。

H3: ローダー戦略の早見表

公式が推奨する解決策は、クエリ発行時にあらかじめ関連データの読み込み方を宣言すること（eager loading）です。.options() に渡します。

H3: コレクションには selectinload、多対一には joinedload

公式ドキュメントの指針は明快です。

ほとんどの場合、selectin loading がコレクションを eager load する最もシンプルで効率的な方法である。

from sqlalchemy.orm import selectinload

# users の取得は1回、addresses の取得も IN句でまとめて1回。計2回で済む
stmt = select(User).options(selectinload(User.addresses))
users = session.scalars(stmt).all()

for user in users:
    print(user.addresses)  # 追加のSQLは飛ばない

一方、多対一（Address.user のような単一参照）には joinedload() が最汎用です。

from sqlalchemy.orm import joinedload

stmt = select(Address).options(joinedload(Address.user))
addresses = session.scalars(stmt).all()

⚠️ joinedload でコレクションを読むときの注意：1 対多・多対多のコレクションに joinedload を使うと、親行が関連件数ぶん重複します。その場合は結果に .unique() を適用する必要がある（公式が明記）。重複と行数の膨張を避けるため、コレクションは原則 selectinload を選ぶべきです。

H3: raiseload() で「うっかり N+1」を本番で禁止する

最も実務的なテクニックがこれです。raiseload() は、明示的に eager load していないリレーションへのアクセスを例外にするガードレールです。

from sqlalchemy.orm import raiseload

# このクエリで取得した User は、addresses に触れると例外になる
stmt = select(User).options(raiseload(User.addresses))

なぜこれが優れているのか？ N+1 は「動いてしまう」ため、テストでもレビューでも見逃されやすく、本番のトラフィック増で初めて顕在化します。raiseload() をデフォルトに据えれば、eager load を書き忘れた瞬間に開発時点で例外として検出でき、N+1 を構造的に「起こり得ないもの」にできます。lazy="raise_on_sql" を使えば、既にセッション内に存在し SQL を要しない多対一アクセスは許可しつつ、新規 SQL を伴う遅延ロードだけを禁止できます。

5. asyncio 対応：FastAPI 時代の非同期データアクセス

FastAPI などの非同期フレームワークでは、AsyncSession を使います。ただし、同期版にはない固有の落とし穴があり、公式ドキュメントもこれを強く警告しています。

H3: 公式に忠実な非同期 ORM の最小構成

import asyncio
from typing import List

from sqlalchemy import ForeignKey, select
from sqlalchemy.ext.asyncio import (
    AsyncAttrs,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    selectinload,
)


class Base(AsyncAttrs, DeclarativeBase):
    """AsyncAttrs を混ぜると await a.awaitable_attrs.<rel> が使えるようになる。"""


class A(Base):
    __tablename__ = "a"
    id: Mapped[int] = mapped_column(primary_key=True)
    bs: Mapped[List["B"]] = relationship()


class B(Base):
    __tablename__ = "b"
    id: Mapped[int] = mapped_column(primary_key=True)
    a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))


async def main() -> None:
    # PostgreSQL の非同期ドライバは asyncpg が定番
    engine = create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test")

    # expire_on_commit=False が非同期では実質必須（理由は後述）
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as session:
        async with session.begin():
            session.add_all([A(bs=[B(), B()]), A(bs=[B()])])

        # 非同期では遅延ロードできないので、selectinload で先読みする
        stmt = select(A).options(selectinload(A.bs))
        result = await session.scalars(stmt)
        for a in result:
            print(a, a.bs)  # ← 先読み済みなので安全

    await engine.dispose()


asyncio.run(main())

H3: 非同期の三大落とし穴（公式の警告）

公式ドキュメントが明記する非同期固有の制約は、知らないと必ず踏みます。

① 暗黙的 IO（遅延ロード）は禁止。 同期版なら a.bs への属性アクセスで暗黙的に SQL が飛びますが、非同期ではこれが MissingGreenlet 系のエラーになります。公式の言葉どおり「属性アクセス時に IO が発生し得るポイントを避けねばならない」。回避策は 3 つ:

1. eager load する：selectinload(A.bs) などをクエリに付ける（最も推奨）。
2. AsyncAttrs を使う：await a.awaitable_attrs.bs で明示的に await する。
3. await session.refresh(a, ["bs"])：属性名を明示してリフレッシュする。

② expire_on_commit=False がほぼ必須。 前述のとおり commit 後はオブジェクトが expire し、次のアクセスで再ロード（=暗黙的 IO）が走ります。非同期ではこの暗黙的 IO 自体が禁止されているため、expire_on_commit=False にして commit 後も属性へ安全にアクセスできるようにします。

③ AsyncSession は並行タスク間で共有不可。 公式は「単一の AsyncSession インスタンスは複数の並行タスクで安全に使えない」と明言しています。asyncio.gather() の各分岐では、必ずタスクごとに別個の AsyncSession を生成してください。

H3: FastAPI での依存性注入パターン

実務では、リクエストごとにセッションを払い出し、レスポンス後に確実にクローズする依存性を定義します。

from collections.abc import AsyncIterator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


async def get_session() -> AsyncIterator[AsyncSession]:
    async with async_session() as session:
        yield session


# ルーターでの利用
@app.get("/users/{user_id}")
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)):
    user = await session.get(User, user_id)
    ...

yield を使った依存性により、ハンドラの処理が成功しても例外で終わっても、async with が必ずセッションをクローズします。回復性とリソースリーク防止を両立する定石です。

6. 本番運用：コネクションプールを正しく設計する

ローカルで動いたコードが本番（特に AWS RDS やサーバーレス）で connection closed / server closed the connection unexpectedly を散発させる——その大半はプール設定の問題です。

create_engine() はデフォルトで QueuePool を統合し、以下のデフォルト値を持ちます。

H3: 本番で必ず設定すべき 2 つのパラメータ

engine = create_engine(
    "postgresql+psycopg2://me@db.example.com/mydb",
    pool_pre_ping=True,   # 切れた接続を掴む事故を防ぐ
    pool_recycle=1800,    # 30分でリサイクル（DB/LB のアイドルタイムアウト対策）
    pool_size=10,
    max_overflow=20,
)

なぜこれが重要なのか？

• pool_pre_ping=True：プールが保持する接続は、DB 側やロードバランサ（RDS Proxy・ELB 等）のアイドルタイムアウトで気付かぬうちに切断されていることがあります。pre_ping は接続を使う直前に軽い ping を打ち、死んでいれば透過的に再接続します。これ一つで stale connection 起因の散発エラーの大半が消えます。
• pool_recycle：MySQL の wait_timeout のように、DB 側が一定時間で接続を切る設定に対し、それより短い秒数で能動的にリサイクルすることで切断を未然に防ぎます。

💡 サーバーレス（AWS Lambda 等）の注意：Lambda は実行環境が頻繁に作り直されるため、大きなプールを各インスタンスが抱えると DB の最大接続数を食い潰します。サーバーレスでは NullPool（プールしない）や、RDS Proxy 等の外部プーラへ寄せる設計が定石です。コスト効率（接続数）と回復性のトレードオフを、実行環境に応じて選択してください。

7. 応用：Repository パターンでデータアクセス層の保守性を最大化する

最後に、これまでの要素を組み合わせ、本番運用に耐える構造へ昇華させます。筆者が経済産業大臣賞を受賞した B2B SaaS で採用した Router → UseCase → Repository → Model の層分離のうち、Repository 層を例に取ります。

ORM のクエリ（select()、ローダー戦略）をルーターやユースケースに直書きすると、ビジネスロジックと永続化の関心が混ざり、SRP（単一責任）が崩れます。Repository はこの境界を作ります。

from collections.abc import Sequence
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload


class UserRepository:
    """User 集約の永続化の関心だけを担う。SQLの知識をここに閉じ込める。"""

    def __init__(self, session: Session) -> None:
        self._session = session

    def get(self, user_id: int) -> User | None:
        # identity map を活用する単純な主キー取得
        return self._session.get(User, user_id)

    def list_with_addresses(self, limit: int = 100) -> Sequence[User]:
        # N+1 を構造的に防ぐ：addresses を必ず selectinload する
        stmt = (
            select(User)
            .options(selectinload(User.addresses))
            .order_by(User.id)
            .limit(limit)
        )
        return self._session.scalars(stmt).all()

    def add(self, user: User) -> None:
        # commit はここでは行わない。トランザクション境界は上位層が握る
        self._session.add(user)

ポイントは、commit() を Repository に置かないことです。トランザクション境界（どこまでを 1 つの作業単位とするか）は、複数の Repository をまたいで調整するユースケース層の責務だからです。

class RegisterUserUseCase:
    def __init__(self, session: Session) -> None:
        self._session = session
        self._users = UserRepository(session)

    def execute(self, name: str, email: str) -> int:
        user = User(name=name, addresses=[Address(email_address=email)])
        self._users.add(user)
        self._session.commit()  # トランザクション境界はユースケースが握る
        return user.id

なぜこの設計が優れているのか？（CLAUDE.md の原則との対応）

• SRP：「SQL をどう書くか（Repository）」と「業務手順とトランザクション境界（UseCase）」が分離され、それぞれ変更理由が 1 つになる。
• ETC / テスト容易性：ユースケースのテストで Repository をモックに差し替えられ、DB なしでビジネスロジックを検証できる。selectinload の追加など永続化の最適化が、上位層に波及しない。
• N+1 の構造的封じ込め：「一覧取得は必ず list_with_addresses を通す」という規約により、ローダー戦略の付け忘れ（=N+1）が設計レベルで起こり得なくなる。
• DRY：同じクエリが複数箇所に散らばらず、Repository のメソッドが唯一の正となる。

これは「速く・安く・安全に」作るための、生成 AI を活用した一人開発でも崩してはならない骨格です。AI に実装を任せる場合でも、層の責務と検証ゲート（型チェック・N+1 検出）を人間が設計することで、スピードと品質を両立できます。

結論：ORM を「動くだけ」から「運用に耐える」へ

SQLAlchemy 2.0 は、Python の型システムと深く統合された、現代的で堅牢な ORM です。本記事で扱った要点を再掲します。

1. Mapped[...] + mapped_column() で、ORM 境界まで型安全を貫徹する。
2. select() 統一 API と session.scalars() へ移行し、レガシーな query() を卒業する。
3. Session のライフサイクルと commit 後 expire を理解し、トランザクション境界をコンテキストマネージャで安全に管理する。
4. selectinload（コレクション）/ joinedload（多対一）/ raiseload（ガード） で N+1 を構造的に根絶する。
5. AsyncSession では暗黙的 IO を避け、expire_on_commit=False と eager load を徹底する。
6. pool_pre_ping / pool_recycle で本番の散発的な接続エラーを防ぐ。
7. Repository パターンで永続化の関心を閉じ込め、保守性・テスト容易性・N+1 耐性を最大化する。

「動くコード」と「10 年運用できるコード」の差は、こうした一つひとつの設計判断の積み重ねにあります。

さらなる探求として、公式ドキュメントの ORM Quickstart、Relationship Loading Techniques、Asynchronous I/O (asyncio) を、本記事の設計観点を念頭に再読することをお勧めします。

本番運用に耐えるバックエンド設計のご相談

筆者は、ここで解説した SQLAlchemy 2.0 のデータアクセス層設計を、経済産業大臣賞を受賞した B2B SaaS の本番環境で実装・運用してきました。型安全・N+1 対策・トランザクション設計・マルチテナントのデータ分離といった、事業の信頼性に直結する基盤を、生成 AI を活用して高速かつ高品質に構築します。Python / PostgreSQL を用いたプロダクト開発・既存システムの改善について、お気軽にご相談ください。