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

> SQLAlchemy 2.0公式ドキュメントに忠実に、Mapped/mapped_columnによる型安全なモデル定義、select()統一API、Sessionのユニットオブワーク、N+1を潰すローダー戦略、asyncio対応、コネクションプール設計、そしてRepositoryパターンによる保守性最大化までを、本番運用の観点で具体的なPythonコードと共に徹底解説します。

- 公開日: 2026-06-24
- 著者: 友田 陽大
- タグ: Python, SQLAlchemy, PostgreSQL, ORM, 型安全, FastAPI, 非同期処理, アーキテクチャ設計
- URL: https://tomodahinata.com/blog/sqlalchemy-2-typed-orm-production-guide

## 要点

- 2.0 では Mapped[...] + mapped_column() で型と NULL 制約を同時に決め、ORM 境界まで型安全を貫徹する
- session.query() を捨て、単一エンティティは session.scalars()、複数カラムは execute() で受ける select() 統一 API へ移行する
- N+1 はコレクションに selectinload、多対一に joinedload、ガードに raiseload を使って構造的に根絶する
- AsyncSession では暗黙的 IO（遅延ロード）が禁止のため、eager load と expire_on_commit=False を徹底する
- 本番の散発的な接続エラーは pool_pre_ping と pool_recycle で防ぎ、Repository パターンで永続化の関心を閉じ込める

---

## **導入：なぜ今「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](https://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」に忠実なモデルを示します。

```python
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 では、**型アノテーション自体がスキーマ情報を持ちます**。

| 書き方 | 生成される列 | 意味 |
| --- | --- | --- |
| `Mapped[int]` | `INTEGER NOT NULL` | 非 Optional は NOT NULL |
| `Mapped[Optional[str]]` | `VARCHAR NULL` | Optional は NULL 許容 |
| `Mapped[str] = mapped_column(String(30))` | `VARCHAR(30) NOT NULL` | 型詳細を `mapped_column` で補足 |

`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` で**一箇所に集約**できます。

```python
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()` 構文**を使えるようになったことです。

```python
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`）** にラップされます。

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

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

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

**実務での指針：**

| やりたいこと | 使うべき API |
| --- | --- |
| 単一エンティティのリストを取得 | `session.scalars(stmt).all()` |
| 主キーで 1 件取得 | `session.get(User, 5)`（identity map を先に見る） |
| 1 件だけ取得（0件/複数件は例外） | `session.scalars(stmt).one()` / `.scalar_one()` |
| 0 件を許容して 1 件 | `session.scalars(stmt).one_or_none()` |
| 複数カラム（`select(User.name, User.id)`） | `session.execute(stmt)` で `Row` タプルを受ける |

`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` とコンテキストマネージャ**

```python
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** が自動で行われます。

```python
# 推奨：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」が罠です。次のコードを見てください。

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

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

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

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

```python
# 対策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"`）** です。次のコードは、一見無害に見えます。

```python
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()` に渡します。

| 戦略 | `lazy=` 相当 | 仕組み | 公式の推奨用途 |
| --- | --- | --- | --- |
| `selectinload()` | `"selectin"` | 親の主キーを `IN` 句でまとめ、追加の 1 回の SELECT で関連行を取得 | **コレクション（1対多・多対多）の第一選択** |
| `joinedload()` | `"joined"` | JOIN で同一結果セットに含める | **多対多ではなく多対一**の最汎用戦略 |
| `subqueryload()` | `"subquery"` | サブクエリで関連行を取得 | ほぼレガシー（`selectinload` に置換済み） |
| `lazyload()` | `"select"` | アクセス時に遅延 SELECT（デフォルト） | N+1 の原因 |
| `raiseload()` | `"raise"` | 遅延ロードを許さず例外を投げる | 本番のガードレール |

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

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

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

```python
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()` が最汎用です。

```python
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 していないリレーションへのアクセスを**例外にする**ガードレールです。

```python
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 の最小構成**

```python
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 での依存性注入パターン**

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

```python
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` を統合し、以下のデフォルト値を持ちます。

| パラメータ | デフォルト | 意味 |
| --- | --- | --- |
| `pool_size` | `5` | 常時保持する接続数 |
| `max_overflow` | `10` | `pool_size` を超えて一時的に作れる追加接続数 |
| `pool_timeout` | `30.0` 秒 | 接続取得を待つ上限 |
| `pool_recycle` | `-1`（無効） | N 秒より古い接続を再生成する |
| `pool_pre_ping` | `False` | チェックアウト時に `SELECT 1` で死活確認する |

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

```python
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 はこの境界を作ります。

```python
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 をまたいで調整するユースケース層の責務だからです。

```python
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](https://docs.sqlalchemy.org/en/20/orm/quickstart.html)、[Relationship Loading Techniques](https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html)、[Asynchronous I/O (asyncio)](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html) を、本記事の設計観点を念頭に再読することをお勧めします。

---

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

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