# marshmallow × Flask × SQLAlchemy で本番REST APIを設計する：境界バリデーションとレスポンス整形

> marshmallow-sqlalchemyのSQLAlchemyAutoSchema・load_instance・auto_fieldを使い、Flask×SQLAlchemyで本番REST APIを設計。load()による入力境界、dump()によるレスポンス整形、ValidationErrorの集約ハンドラで422を返すCRUD、ネスト/リレーション、ページネーション、テストまでを実コードで解説します。

- 公開日: 2026-06-26
- 著者: 友田 陽大
- タグ: Python, marshmallow, Flask, SQLAlchemy, REST API, バリデーション, アーキテクチャ設計
- URL: https://tomodahinata.com/blog/marshmallow-flask-sqlalchemy-rest-api-production-guide

## 要点

- marshmallow-sqlalchemyのSQLAlchemyAutoSchemaで、SQLAlchemyモデルからスキーマを自動生成し、load_instance=Trueでload()がORMインスタンスを返す
- 1つのスキーマがload()で入力境界を、dump()で出力境界を守る。dump_only/load_onlyでマスアサインメントと機密漏洩を防ぐ
- @app.errorhandler(ValidationError)でエラーを一括捕捉し、err.messagesを構造化した422レスポンスに変換する
- ネストしたリレーションはfields.Nestedとinclude_relationshipsで表現し、N+1を意識してdumpする
- pytestでエンドポイントの正常系・異常系を往復検証し、境界の契約をテストで固定する

---

## **導入：REST API には「2 つの境界」がある**

REST API のコードレビューで筆者が最初に見るのは、ビジネスロジックでも DB クエリでもなく、**「外部と接する 2 つの境界がどう守られているか」**です。

- **入口（request → 内部）**：クライアントから来る JSON ボディは、型も値域も信頼できません。`{"role": "admin"}` のような不正なキーが、無防備に ORM モデルへ流れ込めば**権限昇格（マスアサインメント）**になります。
- **出口（内部 → response）**：DB から取り出した ORM モデルには、`password_hash` や内部フラグといった**外に出してはいけない属性**が混ざっています。それをそのまま `jsonify` すれば、機密漏洩です。

この 2 つの境界を、ビュー関数の中に手書きの `if` 文で散らかして守るのは、典型的な技術的負債です。検証漏れはレビューでしか気づけず、出力仕様が変わるたびに複数箇所を直すことになります。

**marshmallow は、この入口と出口を「1 つのスキーマ」で同時に守る**ためのライブラリです。`load()` が入口の検証を、`dump()` が出口の整形を担い、`dump_only` / `load_only` の宣言だけで「書かせない／出さない」をコードの構造として固定します。本記事は、それを **Flask + SQLAlchemy** の本番 REST API として、CRUD・リレーション・ページネーション・テストまで一気通貫で組み上げる実装ガイドです。

筆者は、経済産業大臣賞を受賞した B2B SaaS のバックエンドを **Python / Flask / SQLAlchemy / PostgreSQL** で設計・実装し、`Router → 境界（Schema）→ Model` の層分離で本番運用してきました。ここで示すパターンは、その実戦で磨いたものです。

> 💡 marshmallow 自体の全体像（`Schema` / `fields`、三層バリデーション、`@pre_load` / `@post_load`、3→4 移行）は、対になるピラー記事 [marshmallow 実践ガイド](/blog/marshmallow-python-serialization-validation-production-guide) で詳説しています。本記事はその知識を前提に、**Flask + SQLAlchemy の REST API という具体的な文脈**に落とし込みます。永続化層の設計は [SQLAlchemy 2.0 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide) を併読すると、境界から DB までが一本につながります。

> 💡 **この記事で扱うバージョン**：marshmallow **4.3.0**（2026年4月時点の安定版）／ **marshmallow-sqlalchemy**（SQLAlchemy 2.x 対応版）／ **Flask 3 系**を前提とします。Web 上のサンプルや生成 AI の出力は、いまだ marshmallow 3.x のレガシースタイル（`missing=` / `pass_many=`）が混ざりがちなので注意してください。

---

## **1. アプリ構成：`Router → Schema（境界）→ Model` を層で分ける**

まず全体像です。本記事を通して、**著者（Author）と書籍（Book）の 1 対多**を題材にします。構成は次の 3 層に分離します。

| 層 | 役割 | 本記事での実体 |
| --- | --- | --- |
| Router（ビュー） | HTTP を受け、層を呼び、HTTP を返す | Flask Blueprint |
| Schema（境界） | 入口の検証・出口の整形を一手に担う | marshmallow-sqlalchemy のスキーマ |
| Model（永続化） | テーブル定義と DB アクセス | SQLAlchemy モデル |

依存方向は常に上から下です。**Router は Schema と Model を知ってよいが、Schema と Model は HTTP を一切知らない**——これが「境界の責務」を純粋に保つ鍵です。

### **拡張インスタンスとアプリファクトリ**

循環インポートを避けるため、`db` インスタンスは独立したモジュールに置き、`create_app()` で初期化します（アプリファクトリパターン）。

```python
# extensions.py — 拡張のシングルトンを集約する
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
```

```python
# app.py — アプリファクトリ
from flask import Flask

from extensions import db


def create_app(config: dict | None = None) -> Flask:
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    if config:
        app.config.update(config)

    db.init_app(app)

    # Blueprint とエラーハンドラを登録（後述）
    from api import api_bp, register_error_handlers

    app.register_blueprint(api_bp)
    register_error_handlers(app)

    return app
```

### **SQLAlchemy モデル**

SQLAlchemy 2.x の型付きスタイル（`Mapped` / `mapped_column`）で、Author と Book を定義します。

```python
# models.py
from datetime import datetime, timezone

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from extensions import db


class Author(db.Model):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        default=lambda: datetime.now(timezone.utc)
    )

    # 1 対多：1 人の著者は複数の書籍を持つ
    books: Mapped[list["Book"]] = relationship(
        back_populates="author", cascade="all, delete-orphan"
    )


class Book(db.Model):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    isbn: Mapped[str | None] = mapped_column(String(20), nullable=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"), nullable=False)

    author: Mapped["Author"] = relationship(back_populates="books")
```

このモデルを土台に、次章から marshmallow-sqlalchemy でスキーマを組んでいきます。

---

## **2. `marshmallow-sqlalchemy` 入門：モデルからスキーマを自動生成する**

素の marshmallow では、モデルの各列に対応する `fields.Str()` / `fields.Int()` を手で並べる必要があります。`marshmallow-sqlalchemy` は、**SQLAlchemy モデルの列定義を読み取り、対応するフィールドを自動生成**してくれます。

### **`SQLAlchemyAutoSchema` と `SQLAlchemySchema`**

提供される基底クラスは 2 つあり、「どこまで自動化するか」で使い分けます。

| 基底クラス | フィールド生成 | 使いどころ |
| --- | --- | --- |
| `SQLAlchemyAutoSchema` | `Meta.model` の全列から**自動生成** | モデルとスキーマがほぼ一致する大多数のケース（第一選択） |
| `SQLAlchemySchema` | **明示宣言した `auto_field()` のみ**生成 | 公開する列を厳密に絞りたい、列とフィールドが乖離するケース |

`SQLAlchemyAutoSchema` は「全列を出してから絞る」、`SQLAlchemySchema` は「必要な列だけ足す」——出発点が逆です。本記事は前者を基本に進めます。

### **`class Meta` の主要オプション**

スキーマの挙動は `class Meta` で制御します。marshmallow-sqlalchemy 特有のオプションは次のとおりです。

| `Meta` オプション | 役割 |
| --- | --- |
| `model` | スキーマが対応する SQLAlchemy モデル |
| `load_instance` | `True` で `load()` が dict ではなく**モデルインスタンス**を返す |
| `sqla_session` | `load_instance=True` のとき ORM が使う DB セッション |
| `include_relationships` | `True` でリレーション属性もフィールド化する |
| `include_fk` | `True` で外部キー列（`author_id` 等）も含める |
| `transient` | `True` でセッションに紐付かない一時インスタンスを生成する |

### **`auto_field()`：自動生成を上書きする**

自動生成されたフィールドの一部だけ調整したいときは、`auto_field()` で個別宣言します。これは「列に対応するフィールドを、属性を足して再宣言する」ためのヘルパーです。

```python
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema, auto_field


class AuthorSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True
        include_relationships = True

    # 自動生成された email 列を「必須」として上書き宣言する
    email = auto_field(required=True)
```

`auto_field()` は、用途に応じて 3 つの呼び方ができます。

```python
email = auto_field(required=True)          # 同名の列を上書き
slug = auto_field("url_slug")              # フィールド名と列名が異なる場合
city = auto_field(model=Address)           # 別モデルの列を参照する場合
```

**なぜこれが優れているのか？**
モデルの列定義（`nullable`、`String(120)` など）が、スキーマのフィールドへ自動的に反映されます。列を 1 つ追加してもスキーマを手で直す必要がなく、**モデルとスキーマの二重メンテナンス（DRY 違反）が消えます**。同時に、`auto_field()` で上書きできるため「ほぼ自動・要所だけ手動」という現実的な落としどころが取れます。これは ETC（Easy To Change）の体現です。

---

## **3. 入力境界 `load()`：検証済みの ORM インスタンスを得る**

`load_instance = True` が、REST API での marshmallow-sqlalchemy の主役です。これにより `load()` は、**未知キーの拒否・型検証・業務ルール検証をすべて通過した、検証済みの ORM インスタンス**を返します。`@post_load` を自前で書く必要はありません。

```python
schema = AuthorSchema()
author = schema.load(
    {"name": "友田", "email": "tomoda@example.com"},
    session=db.session,
)
# author は Author インスタンス。そのまま db.session.add(author) できる
```

### **セッションの渡し方は 2 通り**

`load_instance=True` のとき、ORM はインスタンス化に DB セッションを必要とします。渡し方は 2 つです。

```python
# 方法 A：load() の呼び出しごとに渡す（明示的・推奨）
author = AuthorSchema().load(data, session=db.session)


# 方法 B：Meta に固定する
class AuthorSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True
        sqla_session = db.session   # 毎回渡さなくてよい
```

flask-sqlalchemy のスコープ付きセッションでは方法 B も機能しますが、**テストで別セッションへ差し替えやすいのは方法 A** です。本記事はテスト容易性を優先し、方法 A（呼び出し時に `session=` を渡す）で統一します。

### **`required` / `validate` / `unknown=RAISE`**

入口の厳格さは、フィールドと `Meta` で宣言します。

```python
from marshmallow import RAISE, validate
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema, auto_field


class AuthorSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True
        include_relationships = True
        unknown = RAISE   # 未知のキーは ValidationError（既定値だが意図を明示）

    name = auto_field(required=True, validate=validate.Length(min=1, max=120))
    email = auto_field(required=True, validate=validate.Email())
```

`unknown = RAISE` は marshmallow の既定値ですが、**REST API の入口では明示しておく**ことを勧めます。「未知のキーは黙って捨てるのではなく、エラーにする」という入口の契約を、スキーマを読むだけで把握できるからです。これにより `{"name": "...", "is_admin": true}` のような余計なキーは、検証段階で弾かれます。

**なぜこれが優れているのか？**
ビュー関数に「`name` は必須か」「`email` は形式が正しいか」を判定する `if` 文が一切要りません。境界を通過した `author` は「検証済みの ORM インスタンス」であることがコード上で保証され、下流はその前提を信頼できます。検証ロジックがビジネスロジックに混入しない——SRP（単一責任）の徹底です。

---

## **4. セキュリティ：`dump_only` / `load_only` で 2 つの事故を構造的に防ぐ**

ここが REST API 設計で最も事故が起きやすい箇所です。marshmallow は、**「書かせない」「出さない」をスキーマの宣言で固定**できます。

### **`dump_only`：マスアサインメントを防ぐ**

`id` / `created_at` のようなサーバーが決める値や、`role` のような権限フィールドを `request` から無防備に受け取ると、攻撃者が `{"role": "admin"}` を送り込む**マスアサインメント脆弱性**になります。`dump_only=True` のフィールドは**出力専用**で、`load()` では完全に無視されます。

```python
class AuthorSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True
        include_relationships = True
        unknown = RAISE

    id = auto_field(dump_only=True)          # サーバーが採番。load では書けない
    created_at = auto_field(dump_only=True)  # 生成時刻はサーバーが決める
    name = auto_field(required=True, validate=validate.Length(min=1, max=120))
    email = auto_field(required=True, validate=validate.Email())
```

> ⚠️ **マスアサインメントの罠**：`SQLAlchemyAutoSchema` は既定でモデルの全列をフィールド化します。つまり **`role` や `is_admin` のような特権列も、何もしなければ `load()` で書き込めてしまう**ということです。OWASP が「Mass Assignment」として警告するこの脆弱性は、`request.get_json()` を ORM へ素通しするコードで頻発します。サーバーが決定すべき列には**必ず `dump_only=True` を付ける**——これを設計の既定ルールにしてください。新しい特権列を追加したら、まず `dump_only` を検討するのが鉄則です。

### **`load_only`：機密の漏洩を防ぐ**

パスワードやトークンを、誤ってレスポンスに含めるのは典型的な情報漏洩です。`load_only=True` のフィールドは**入力専用**で、`dump()` の出力には決して含まれません。

```python
from marshmallow import fields, validate


class SignupSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True

    # password は load では受け取るが、dump では絶対に出力されない
    password = fields.Str(load_only=True, required=True, validate=validate.Length(min=12))
```

| 宣言 | `load()`（入口） | `dump()`（出口） | 主な用途 |
| --- | --- | --- | --- |
| `dump_only=True` | 無視される（書けない） | 出力される | `id` / `created_at` / `role` |
| `load_only=True` | 受け取る | 出力されない（漏れない） | `password` / `token` |
| 既定（どちらも無し） | 受け取る | 出力される | 通常の属性 |

**なぜこれが優れているのか？**
セキュリティを「レビューでの気づき」に頼ると、必ずいつか漏れます。`dump_only` / `load_only` は、**スキーマ定義そのものに安全制約を宣言**することで、「うっかり書ける／うっかり出る」を**構造的に不可能**にします。これは CLAUDE.md のセキュリティ原則「すべての外部入力を境界で検証し、最小権限を適用する」を、運用ではなくコードの構造で実装したものです。

---

## **5. 出力境界 `dump()`：レスポンスをビューごとに整形する**

出口は `dump()` が担います。`load_only` で機密が落ちることはすでに保証済みなので、ここでは**「どの列を、どの粒度で見せるか」**を制御します。

### **`only` / `exclude` でビューを切り替える**

同じ `AuthorSchema` から、`only` / `exclude` を切り替えるだけで「一覧用（軽量）」「詳細用（全項目）」を作り分けられます。ビューごとにスキーマを増やす必要はありません。

```python
detail_schema = AuthorSchema()                       # 全項目
list_schema = AuthorSchema(only=("id", "name"))      # 一覧は id と name だけ
public_schema = AuthorSchema(exclude=("email",))     # 公開ビューはメールを隠す
```

### **`@post_dump` でエンベロープに包む**

API レスポンスを `{"data": ...}` という共通の封筒で包むと、メタ情報（ページング等）を足す余地が生まれ、クライアントの取り回しが安定します。`@post_dump` を使い、コレクションかどうかは **v4 の `pass_collection=True`** で受け取ります（3.x の `pass_many` から改名）。

```python
from marshmallow import post_dump


class AuthorSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True
        include_relationships = True
        unknown = RAISE

    id = auto_field(dump_only=True)
    created_at = auto_field(dump_only=True)
    name = auto_field(required=True, validate=validate.Length(min=1, max=120))
    email = auto_field(required=True, validate=validate.Email())

    @post_dump(pass_collection=True)
    def wrap(self, data, many, **kwargs):
        # many=True（一覧）でも many=False（単体）でも data キーで統一する
        return {"data": data}
```

> 💡 エンベロープを**全スキーマで統一したい**場合は、`@post_dump` を持つ基底スキーマを 1 つ定義し、各スキーマがそれを継承する設計が有効です。封筒の形（`data` / `meta` のキー名）が 1 箇所に集約され、API 全体のレスポンス形が一貫します。ただしページネーションの `meta` を足す一覧系は、後述（10 章）のようにビュー側で組む方が見通しが良い場合もあります。設計の落としどころはチームの規約に合わせてください。

**なぜこれが優れているのか？**
「一覧は軽く、詳細は厚く」というプレゼンテーションの差を、**単一のスキーマ定義から `only` / `exclude` の引数だけで**表現できます。ビューが増えるたびにモデルやシリアライザを増やす必要がなく、出力仕様の変更がスキーマ 1 箇所に局所化されます。marshmallow が「レスポンス整形に強い」と言われる核心です。

---

## **6. エラーの一元化：`@app.errorhandler(ValidationError)` で 422 を返す**

各ルートに `try / except ValidationError` を書くと、同じ変換ロジックがエンドポイントの数だけ重複します。これは典型的な DRY 違反です。Flask のエラーハンドラを使えば、**`ValidationError` を 1 箇所で捕捉し、構造化した 422 レスポンスへ変換**できます。

```python
# api.py — エラーハンドラの集約登録
from flask import Flask, jsonify
from marshmallow import ValidationError


def register_error_handlers(app: Flask) -> None:
    @app.errorhandler(ValidationError)
    def handle_validation_error(err: ValidationError):
        # err.messages は {フィールド名: [メッセージ, ...]} の dict
        return jsonify(errors=err.messages), 422

    @app.errorhandler(404)
    def handle_not_found(err):
        return jsonify(errors={"_schema": ["リソースが見つかりません。"]}), 404
```

これで、ルート側は `schema.load(...)` をそのまま呼ぶだけでよくなります。検証に失敗すれば `ValidationError` が送出され、ハンドラが一括で 422 に変換します。

```python
# ルート側：try/except は不要。失敗すれば errorhandler が拾う
@api_bp.post("/authors")
def create_author():
    author = AuthorSchema().load(request.get_json(), session=db.session)
    db.session.add(author)
    db.session.commit()
    return jsonify(AuthorSchema().dump(author)), 201
```

422（Unprocessable Entity）は「構文は正しいが、意味的に処理できない」エラーを表す標準的なステータスで、バリデーション失敗の表現として広く使われます。`err.messages` はフィールド名をキーにした dict なので、フロントエンドはどの入力欄にエラーを出すかを機械的に判定できます。

**なぜこれが優れているのか？**
「検証エラーを HTTP レスポンスへ変換する」という知識が、アプリ全体で**ただ 1 箇所**に存在します。新しいエンドポイントを追加しても、エラー整形のコードを書く必要は一切ありません。`try / except` の重複が消え、ルート関数には本質的な処理（保存する・取り出す）だけが残ります——DRY と SRP の同時達成です。

---

## **7. CRUD 全実装：1 つの Blueprint に I/O 境界を集約する**

ここまでの部品を組み合わせ、Author の CRUD を完成させます。各ルートは驚くほど薄くなります——検証も整形もスキーマに委譲済みだからです。

```python
# api.py（続き）
from flask import Blueprint, jsonify, request

from extensions import db
from models import Author
from schemas import AuthorSchema

api_bp = Blueprint("api", __name__, url_prefix="/api")


# CREATE — 201 / 422
@api_bp.post("/authors")
def create_author():
    # ① 入口：未知キー拒否・型検証・業務ルールを通過した ORM インスタンスだけが残る
    author = AuthorSchema().load(request.get_json(), session=db.session)
    db.session.add(author)
    db.session.commit()
    # ② 出口：dump_only / load_only により安全に整形された表現だけが外へ出る
    return jsonify(AuthorSchema().dump(author)), 201


# LIST — 200（many=True）
@api_bp.get("/authors")
def list_authors():
    authors = db.session.scalars(db.select(Author)).all()
    # 一覧は軽量ビュー。many=True で配列としてダンプする
    return jsonify(AuthorSchema(many=True, only=("id", "name")).dump(authors)), 200


# DETAIL — 200 / 404
@api_bp.get("/authors/<int:author_id>")
def get_author(author_id: int):
    # get_or_404 は対象が無ければ 404 を送出（errorhandler が JSON 化）
    author = db.get_or_404(Author, author_id)
    return jsonify(AuthorSchema().dump(author)), 200


# UPDATE（部分更新） — 200 / 422 / 404
@api_bp.patch("/authors/<int:author_id>")
def update_author(author_id: int):
    author = db.get_or_404(Author, author_id)
    # instance= で既存インスタンスへ上書きロード、partial=True で未送信の required を免除
    author = AuthorSchema().load(
        request.get_json(),
        session=db.session,
        instance=author,
        partial=True,
    )
    db.session.commit()
    return jsonify(AuthorSchema().dump(author)), 200


# DELETE — 204 / 404
@api_bp.delete("/authors/<int:author_id>")
def delete_author(author_id: int):
    author = db.get_or_404(Author, author_id)
    db.session.delete(author)
    db.session.commit()
    return "", 204
```

PATCH の 2 点が肝です。`instance=author` を渡すと、`load_instance=True` のスキーマは**新しいインスタンスを作らず既存インスタンスへ上書き**します。そして `partial=True` により、送られなかったフィールドの `required` 検証はスキップされ、「送った分だけ更新」という PATCH の意味論が成立します。

> 💡 上のエンベロープ章（5 章）で `@post_dump` を有効にしている場合、`dump()` の戻りは `{"data": ...}` になります。CRUD の整合のため、本章のスキーマでは封筒を外すか、レスポンス全体を `{"data": ...}` に統一するかを**プロジェクトで 1 つに決めて**ください。混在が最も保守を難しくします。

---

## **8. ネストとリレーション：`fields.Nested` と `include_relationships`**

Book は Author に属します。このリレーションをレスポンスでどう表現するかは、`include_relationships` と `fields.Nested` で制御します。

### **読み取り：リレーションをネストして出す**

`include_relationships = True` だけでも関連は出力されますが、既定では**主キーの配列**になりがちです。「関連オブジェクトの中身」を出したいなら、`fields.Nested` で明示的にスキーマを入れ子にします。

```python
# schemas.py
from marshmallow import RAISE, fields, validate
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema, auto_field

from models import Author, Book


class BookSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Book
        load_instance = True
        include_fk = True        # author_id（外部キー）を入出力に含める
        unknown = RAISE

    id = auto_field(dump_only=True)
    title = auto_field(required=True, validate=validate.Length(min=1, max=200))


class AuthorSchema(SQLAlchemyAutoSchema):
    class Meta:
        model = Author
        load_instance = True
        include_relationships = True
        unknown = RAISE

    id = auto_field(dump_only=True)
    created_at = auto_field(dump_only=True)
    name = auto_field(required=True, validate=validate.Length(min=1, max=120))
    email = auto_field(required=True, validate=validate.Email())

    # 著者に紐づく書籍を、タイトルだけのネストとして出力する
    books = fields.Nested(BookSchema, many=True, only=("id", "title"), dump_only=True)
```

詳細レスポンスは、著者と書籍がネストした構造になります。

```json
{
  "id": 1,
  "name": "友田",
  "email": "tomoda@example.com",
  "created_at": "2026-06-26T09:00:00+00:00",
  "books": [
    { "id": 10, "title": "境界の設計" },
    { "id": 11, "title": "型安全な API" }
  ]
}
```

### **書き込み：外部キーで関連付ける**

書籍の作成では、`include_fk=True` により `author_id` を入力として受け取り、関連を張ります。

```python
@api_bp.post("/books")
def create_book():
    # {"title": "...", "author_id": 1} を受け取り、検証して保存する
    book = BookSchema().load(request.get_json(), session=db.session)
    db.session.add(book)
    db.session.commit()
    return jsonify(BookSchema().dump(book)), 201
```

### **N+1 に注意する**

> ⚠️ **N+1 クエリの罠**：著者一覧を `books` ネスト付きで `dump()` すると、marshmallow は**各著者ごとに `books` リレーションへアクセス**します。リレーションが遅延ロード（lazy）のままだと、**著者 N 件に対して書籍取得クエリが N 回**追加で飛び、件数に比例してレスポンスが遅くなります。これはアプリが遅くなる古典的かつ最頻出の原因です。対策は、ダンプ前に関連を**まとめて先読み**することです。

```python
from sqlalchemy.orm import selectinload

# selectinload で books を 1 回の追加クエリでまとめて取得（N+1 を回避）
stmt = db.select(Author).options(selectinload(Author.books))
authors = db.session.scalars(stmt).all()
return jsonify(AuthorSchema(many=True).dump(authors)), 200
```

`selectinload`（または `joinedload`）で関連を先読みすれば、クエリ数は件数に依存せず一定に保たれます。**「marshmallow が遅い」のではなく、ダンプ時にトリガーされる遅延ロードが遅い**——この区別が、本番性能を守る上で決定的です。SQLAlchemy 側のローディング戦略の詳細は [SQLAlchemy 2.0 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide) を参照してください。

**なぜこれが優れているのか？**
関連の「読み取り表現（ネストの粒度）」と「書き込み手段（外部キー）」を、同じスキーマ群の中で宣言的に分離できます。`only=("id", "title")` を変えるだけで埋め込みの厚みを調整でき、関連を増やしてもビュー関数は変わりません。リレーションの表現がスキーマに局所化される——ETC の実践です。

---

## **9. ページネーション：エンベロープと `meta` で一貫したレスポンス形に**

一覧 API は、件数が増えれば必ずページングが要ります。flask-sqlalchemy の `paginate()` を使い、データを `data`、ページ情報を `meta` に分けた一貫したレスポンス形を組みます。

```python
@api_bp.get("/authors")
def list_authors():
    page = request.args.get("page", default=1, type=int)
    per_page = request.args.get("per_page", default=20, type=int)
    # per_page に上限を設けて、過大なリクエストによる負荷を防ぐ
    per_page = min(per_page, 100)

    # N+1 回避のため books を先読みしてからページングする
    stmt = db.select(Author).options(selectinload(Author.books)).order_by(Author.id)
    pagination = db.paginate(stmt, page=page, per_page=per_page, error_out=False)

    body = {
        "data": AuthorSchema(many=True, only=("id", "name")).dump(pagination.items),
        "meta": {
            "page": pagination.page,
            "per_page": pagination.per_page,
            "total": pagination.total,
            "pages": pagination.pages,
        },
    }
    return jsonify(body), 200
```

`error_out=False` により、存在しないページ番号でも 404 を投げず**空配列**を返します（クライアントが末尾を超えてページ送りしても安全）。`per_page` に上限（ここでは 100）を設けるのは、`per_page=100000` のような過大なリクエストで DB とメモリを枯渇させる**サービス拒否（DoS）対策**です。境界で外部入力を検証する原則は、ページングのパラメータにも等しく適用します。

レスポンスは次の形に統一されます。

```json
{
  "data": [
    { "id": 1, "name": "友田" },
    { "id": 2, "name": "山田" }
  ],
  "meta": { "page": 1, "per_page": 20, "total": 57, "pages": 3 }
}
```

**なぜこれが優れているのか？**
すべての一覧 API が `{"data": [...], "meta": {...}}` という同じ封筒を返すため、クライアントは 1 種類のパース処理を使い回せます。`total` / `pages` がレスポンスに含まれることで、ページャ UI をクライアント側で組み立てられます。レスポンス形の一貫性は、API の使いやすさそのものです。

---

## **10. テスト：境界の契約を pytest で固定する**

境界の振る舞い——「正しい入力は 201、不正な入力は 422、存在しないリソースは 404」——は、API の**契約**です。契約はテストで固定して初めて、リファクタや機能追加で壊れないことを保証できます。

### **フィクスチャ：テスト用アプリと DB を用意する**

```python
# conftest.py
import pytest

from app import create_app
from extensions import db


@pytest.fixture()
def app():
    # インメモリ SQLite でテストごとにクリーンな DB を作る
    app = create_app({"SQLALCHEMY_DATABASE_URI": "sqlite:///:memory:", "TESTING": True})
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()


@pytest.fixture()
def client(app):
    return app.test_client()
```

### **エンドポイントの往復検証：201 / 422 / 404**

```python
# test_authors.py
def test_create_author_returns_201(client):
    res = client.post("/api/authors", json={"name": "友田", "email": "a@example.com"})
    assert res.status_code == 201
    body = res.get_json()
    assert body["name"] == "友田"
    assert "id" in body            # サーバーが採番した id が出力される
    assert "password" not in body  # load_only の機密は出力に含まれない


def test_create_author_rejects_invalid_email_with_422(client):
    res = client.post("/api/authors", json={"name": "友田", "email": "not-an-email"})
    assert res.status_code == 422
    # errorhandler が err.messages を errors キーへ整形している
    assert "email" in res.get_json()["errors"]


def test_create_author_rejects_unknown_key_with_422(client):
    # unknown=RAISE により、未知のキー（role）はマスアサインメントとして弾かれる
    res = client.post(
        "/api/authors",
        json={"name": "友田", "email": "a@example.com", "role": "admin"},
    )
    assert res.status_code == 422
    assert "role" in res.get_json()["errors"]


def test_get_missing_author_returns_404(client):
    res = client.get("/api/authors/9999")
    assert res.status_code == 404


def test_patch_updates_partial_fields(client):
    created = client.post(
        "/api/authors", json={"name": "友田", "email": "a@example.com"}
    ).get_json()
    # name だけ送る。partial=True により email の required は免除される
    res = client.patch(f"/api/authors/{created['id']}", json={"name": "山田"})
    assert res.status_code == 200
    assert res.get_json()["name"] == "山田"
```

### **スキーマ単体テスト：ビューを介さず境界を直接検証する**

エンドポイントを介さず、スキーマの `load()` / `dump()` を直接叩くテストも価値があります。検証ロジックの単体テストとして高速で、原因の切り分けが容易です。

```python
# test_schemas.py
import pytest
from marshmallow import ValidationError

from schemas import AuthorSchema


def test_load_rejects_mass_assignment_of_dump_only_field(app):
    with app.app_context():
        from extensions import db

        # dump_only の id を送り込んでも、load では黙殺される（書き込まれない）
        author = AuthorSchema().load(
            {"id": 999, "name": "友田", "email": "a@example.com"},
            session=db.session,
        )
        assert author.id is None  # クライアント指定の id は反映されない


def test_load_raises_for_missing_required_field(app):
    with app.app_context():
        from extensions import db

        with pytest.raises(ValidationError) as exc:
            AuthorSchema().load({"name": "友田"}, session=db.session)  # email 欠落
        assert "email" in exc.value.messages
```

**なぜこれが優れているのか？**
「不正入力は 422、機密は出力されない、`dump_only` は書けない」という**セキュリティ上の契約がテストで固定**されます。将来スキーマをリファクタしても、`dump_only` を外してしまうような退行はテストが即座に検出します。境界の安全性は、宣言（スキーマ）とテスト（契約の固定）の二重で守られます。

---

## **結論：1 つのスキーマで、入口と出口を同時に設計する**

Flask + SQLAlchemy の REST API において、marshmallow-sqlalchemy は「境界の門番」を最小のボイラープレートで実現します。本記事の要点を再掲します。

1. **`SQLAlchemyAutoSchema` + `auto_field()`** で、SQLAlchemy モデルからスキーマを自動生成し、モデルとスキーマの二重メンテナンスを消す。
2. **`load_instance = True`** で `load()` が検証済みの ORM インスタンスを返し、`session=` を渡してそのまま `add()` できる。
3. **`dump_only`（マスアサインメント防止）／ `load_only`（機密漏洩防止）／ `unknown=RAISE`（未知キー拒否）** で、2 大セキュリティ事故を**スキーマの構造**で封じる。
4. **`@app.errorhandler(ValidationError)`** で `err.messages` を 422 へ一括変換し、各ルートから手書きの `try / except` を排除する（DRY）。
5. **`fields.Nested` / `include_relationships`** でリレーションを表現し、`selectinload` で **N+1 を回避**してからダンプする。
6. **`@post_dump` のエンベロープ + `meta`** でレスポンス形を統一し、**pytest で 201 / 422 / 404 を往復検証**して境界の契約を固定する。

「動く REST API」と「10 年運用できる REST API」を分けるのは、フレームワークの選定でも ORM の速さでもなく、**入口と出口という 2 つの境界を、どれだけ宣言的に・テスト可能に守れているか**です。marshmallow-sqlalchemy は、その境界を 1 つのスキーマとして表現する、実績ある道具です。

さらなる探求として、公式ドキュメントの以下を本記事の設計観点を念頭に再読することをお勧めします。

- [marshmallow-sqlalchemy 公式ドキュメント](https://marshmallow-sqlalchemy.readthedocs.io/en/latest/)
- [marshmallow: Nesting Schemas](https://marshmallow.readthedocs.io/en/stable/nesting.html)
- [marshmallow: Extending Schemas（pre/post 処理）](https://marshmallow.readthedocs.io/en/stable/extending/schema_validation.html)
- [SQLAlchemy: Relationship Loading Techniques（N+1 対策）](https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html)

---

### **型安全なバックエンド設計のご相談**

筆者は、ここで解説した「REST API の入口で外部入力を必ず検証し、出口で機密を落として安全に整形する」という境界設計を、経済産業大臣賞を受賞した B2B SaaS の本番環境で、Flask / SQLAlchemy / marshmallow のスタックとして実装・運用してきました。マスアサインメント対策・レスポンス整形・ORM 連携・N+1 を踏まないリレーション設計・契約を固定するテストまで、**事業の信頼性に直結する API 基盤**を、生成 AI を活用して高速かつ高品質に構築します。Python を用いた REST API 開発・既存 API の堅牢化について、お気軽にご相談ください。