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

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

db = SQLAlchemy()

# 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

# 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")

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)

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

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

# 方法 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   # 毎回渡さなくてよい

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())

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())

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))

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

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}

# 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

# ルート側：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

# 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

# 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)

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

@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

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

@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

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

# 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()

# 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"] == "山田"

# 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