Skip to main content
友田 陽大
marshmallow
Python
marshmallow
Flask
SQLAlchemy
REST API
バリデーション
アーキテクチャ設計

Designing a Production REST API with marshmallow × Flask × SQLAlchemy: Boundary Validation and Response Shaping

Using marshmallow-sqlalchemy's SQLAlchemyAutoSchema, load_instance, and auto_field, design a production REST API with Flask × SQLAlchemy. We explain — with real code — the input boundary via load(), response shaping via dump(), CRUD that returns 422 through an aggregated ValidationError handler, nesting/relationships, pagination, and testing.

Published
Reading time
21 min read
Author
友田 陽大
Share
Contents

Introduction: A REST API Has "Two Boundaries"

In a REST API code review, the first thing I look at is neither the business logic nor the DB queries, but "how the two boundaries that touch the outside world are guarded."

  • Entrance (request → internal): the JSON body coming from the client is untrustworthy in both type and value range. If an illegitimate key like {"role": "admin"} flows defenselessly into an ORM model, it becomes privilege escalation (mass assignment).
  • Exit (internal → response): an ORM model pulled from the DB has attributes that must not go out, like password_hash or internal flags, mixed in. jsonify it as-is and that's a secret leak.

Guarding these two boundaries with hand-written if statements scattered through view functions is textbook technical debt. Validation gaps are caught only in review, and every time the output spec changes you end up fixing multiple places.

marshmallow is a library for guarding this entrance and exit simultaneously with "a single schema." load() handles entrance validation, dump() handles exit shaping, and just by declaring dump_only / load_only you pin "don't let it be written / don't let it out" as a structural property of the code. This article is an implementation guide that builds it all the way through — CRUD, relationships, pagination, and testing — as a production REST API on Flask + SQLAlchemy.

I designed and built the backend of an economic-ministry-award-winning B2B SaaS in Python / Flask / SQLAlchemy / PostgreSQL, running it in production with a Router → boundary (Schema) → Model layer separation. The patterns shown here are honed in that real-world battle.

💡 The overall picture of marshmallow itself (Schema / fields, the three-layer validation, @pre_load / @post_load, the 3→4 migration) is detailed in the paired pillar article marshmallow Practical Guide. This article takes that knowledge as a premise and drops it into the concrete context of a Flask + SQLAlchemy REST API. Read the persistence-layer design alongside SQLAlchemy 2.0 Practical Guide and you'll have a single thread running from the boundary all the way to the DB.

💡 Versions covered in this article: marshmallow 4.3.0 (the stable version as of April 2026) / marshmallow-sqlalchemy (the SQLAlchemy 2.x-compatible version) / Flask 3 line. Note that web samples and generative-AI output still tend to mix in marshmallow 3.x legacy style (missing= / pass_many=).


1. App Structure: Separate Router → Schema (boundary) → Model by Layer

First, the overall picture. Throughout this article we'll use the one-to-many of Author and Book as the theme. The structure separates into the following three layers.

LayerRoleConcrete form in this article
Router (view)Receives HTTP, calls the layers, returns HTTPFlask Blueprint
Schema (boundary)Single-handedly handles entrance validation and exit shapingmarshmallow-sqlalchemy schema
Model (persistence)Table definition and DB accessSQLAlchemy model

The dependency direction is always top-down. The Router may know the Schema and the Model, but the Schema and the Model know nothing of HTTP — this is the key to keeping the "boundary's responsibility" pure.

Extension Instances and the App Factory

To avoid circular imports, place the db instance in an independent module and initialize it in create_app() (the app factory pattern).

# 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

SQLAlchemy Models

In SQLAlchemy 2.x's typed style (Mapped / mapped_column), define Author and Book.

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

Building on this model, from the next section we'll assemble schemas with marshmallow-sqlalchemy.


2. marshmallow-sqlalchemy Primer: Auto-Generating a Schema from a Model

With plain marshmallow, you have to line up fields.Str() / fields.Int() by hand for each column of the model. marshmallow-sqlalchemy reads the SQLAlchemy model's column definitions and auto-generates the corresponding fields.

SQLAlchemyAutoSchema and SQLAlchemySchema

There are two base classes provided, used differently by "how far you automate."

Base classField generationWhere to use
SQLAlchemyAutoSchemaAuto-generated from all columns of Meta.modelThe vast majority of cases where the model and schema nearly match (first choice)
SQLAlchemySchemaOnly the explicitly declared auto_field() are generatedWhen you want to strictly narrow the exposed columns, or when columns and fields diverge

SQLAlchemyAutoSchema is "emit all columns, then narrow"; SQLAlchemySchema is "add only the columns you need" — their starting points are opposite. This article proceeds with the former as the base.

Key class Meta Options

A schema's behavior is controlled by class Meta. The options specific to marshmallow-sqlalchemy are as follows.

Meta optionRole
modelThe SQLAlchemy model the schema corresponds to
load_instanceWith True, load() returns a model instance instead of a dict
sqla_sessionThe DB session the ORM uses when load_instance=True
include_relationshipsWith True, relationship attributes are also turned into fields
include_fkWith True, foreign-key columns (author_id, etc.) are also included
transientWith True, generates a transient instance not bound to a session

auto_field(): Overriding Auto-Generation

When you want to tweak just part of an auto-generated field, declare it individually with auto_field(). This is a helper for "re-declaring the field corresponding to a column, with added attributes."

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() can be called in three ways, depending on the use.

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

Why is this superior? The model's column definitions (nullable, String(120), etc.) are automatically reflected into the schema's fields. Add one column and you don't need to hand-fix the schema — the double maintenance of model and schema (a DRY violation) disappears. At the same time, because you can override with auto_field(), you get the realistic landing point of "mostly automatic, manual only where it matters." This embodies ETC (Easy To Change).


3. The Input Boundary load(): Get a Validated ORM Instance

load_instance = True is the star of marshmallow-sqlalchemy in a REST API. With it, load() returns a validated ORM instance that has passed unknown-key rejection, type validation, and business-rule validation in full. You don't need to write your own @post_load.

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

There Are Two Ways to Pass the Session

When load_instance=True, the ORM needs a DB session to instantiate. There are two ways to pass it.

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

With flask-sqlalchemy's scoped session, method B also works, but method A is easier to swap to a separate session in tests. This article prioritizes testability and standardizes on method A (pass session= at call time).

required / validate / unknown=RAISE

The strictness of the entrance is declared in the fields and Meta.

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 is marshmallow's default, but I recommend stating it explicitly at the entrance of a REST API. The reason is that you can grasp the entrance's contract — "an unknown key is not silently dropped but errored" — just by reading the schema. This rejects extraneous keys like {"name": "...", "is_admin": true} at the validation stage.

Why is this superior? The view function needs no if statements at all to judge "is name required?" or "is email in a valid format?" The author that has passed the boundary is guaranteed in code to be "a validated ORM instance," and downstream can trust that premise. Validation logic doesn't bleed into business logic — a thorough application of SRP (single responsibility).


4. Security: Structurally Prevent Two Accidents with dump_only / load_only

This is where accidents most easily happen in REST API design. marshmallow lets you pin "don't let it be written" / "don't let it out" via schema declarations.

dump_only: Preventing Mass Assignment

Server-decided values like id / created_at, or privilege fields like role, received defenselessly from the request, become a mass-assignment vulnerability where an attacker sends in {"role": "admin"}. A dump_only=True field is output-only and is completely ignored in load().

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

⚠️ The mass-assignment trap: SQLAlchemyAutoSchema turns all of the model's columns into fields by default. That means even privileged columns like role or is_admin, if you do nothing, become writable in load(). This vulnerability, which OWASP warns about as "Mass Assignment," frequently occurs in code that passes request.get_json() straight through to the ORM. For columns the server should decide, always add dump_only=True — make this the default rule of your design. When you add a new privileged column, the iron rule is to consider dump_only first.

load_only: Preventing Secret Leakage

Accidentally including a password or token in the response is a textbook information leak. A load_only=True field is input-only and is never included in dump()'s output.

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))
Declarationload() (entrance)dump() (exit)Main use
dump_only=TrueIgnored (not writable)Outputid / created_at / role
load_only=TrueReceivedNot output (doesn't leak)password / token
Default (neither)ReceivedOutputOrdinary attributes

Why is this superior? Rely on security being "noticed in review" and it will leak someday, guaranteed. dump_only / load_only declare the safety constraints in the schema definition itself, making "accidentally writable / accidentally exposed" structurally impossible. This implements CLAUDE.md's security principle "validate all external input at the boundary and apply least privilege," not as an operational practice but as a structure of the code.


5. The Output Boundary dump(): Shape Responses Per View

The exit is handled by dump(). Since it's already guaranteed that load_only drops secrets, here we control "which columns to show, at what granularity."

Switch Views with only / exclude

From the same AuthorSchema, just by switching only / exclude, you can create "for-list (lightweight)" and "for-detail (all fields)." There's no need to add a schema per view.

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

Wrap in an Envelope with @post_dump

Wrapping the API response in a common envelope {"data": ...} creates room to add meta info (paging, etc.) and stabilizes the client's handling. Use @post_dump, and receive whether it's a collection via v4's pass_collection=True (renamed from 3.x's pass_many).

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}

💡 If you want to unify the envelope across all schemas, defining a single base schema with @post_dump and having each schema inherit it is effective. The envelope's shape (the key names data / meta) is consolidated in one place, and the API's overall response shape stays consistent. That said, for list endpoints that add a pagination meta, it can be clearer to assemble it on the view side as shown later (chapter 10). Match the design's landing point to your team's conventions.

Why is this superior? You can express the presentation difference of "lists light, details thick" from a single schema definition, with just the only / exclude arguments. There's no need to add a model or serializer every time a view increases, and changes to the output spec are localized to one place in the schema. This is the core of why marshmallow is said to be "strong at response shaping."


6. Centralizing Errors: Return 422 with @app.errorhandler(ValidationError)

Write try / except ValidationError in each route and the same conversion logic duplicates across the number of endpoints. This is a textbook DRY violation. Using Flask's error handler, you can catch ValidationError in one place and convert it into a structured 422 response.

# 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

With this, the route side only needs to call schema.load(...) as-is. If validation fails, a ValidationError is raised and the handler converts it to 422 in one place.

# ルート側: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) is the standard status representing "syntactically correct but semantically unprocessable" errors, widely used as the expression of a validation failure. Since err.messages is a dict keyed by field name, the frontend can mechanically determine which input field to show the error on.

Why is this superior? The knowledge "convert a validation error into an HTTP response" exists in exactly one place across the whole app. Even when you add a new endpoint, you never need to write error-shaping code. The duplication of try / except disappears, and only the essential work (save, fetch) remains in the route functions — DRY and SRP achieved simultaneously.


7. Full CRUD Implementation: Consolidating the I/O Boundary in One Blueprint

Combining the parts so far, let's complete the Author CRUD. Each route becomes surprisingly thin — because validation and shaping are already delegated to the schema.

# 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

The two key points of PATCH. Passing instance=author makes a load_instance=True schema overwrite the existing instance rather than create a new one. And partial=True skips required validation for fields that weren't sent, so the PATCH semantics of "update only what you sent" hold.

💡 If you have @post_dump enabled per the envelope chapter (chapter 5) above, dump()'s return becomes {"data": ...}. For CRUD consistency, in this chapter's schema, decide in one place for the project whether to remove the envelope or unify the entire response to {"data": ...}. Mixing the two is what makes maintenance hardest.


8. Nesting and Relationships: fields.Nested and include_relationships

Book belongs to Author. How to express this relationship in the response is controlled by include_relationships and fields.Nested.

Reading: Output the Relationship Nested

include_relationships = True alone does output the relationship, but by default it tends to be an array of primary keys. If you want "the contents of the related object," nest the schema explicitly with fields.Nested.

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

The detail response becomes a structure with the author and books nested.

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

Writing: Associate via the Foreign Key

When creating a book, include_fk=True lets you receive author_id as input and establish the association.

@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

Watch Out for N+1

⚠️ The N+1 query trap: dump() an author list with the books nest, and marshmallow accesses the books relationship for each author. If the relationship stays lazy-loaded (lazy), a book-fetch query fires N additional times for N authors, and the response slows in proportion to the count. This is the classic, most frequent cause of an app being slow. The countermeasure is to pre-fetch the relationships in bulk before dumping.

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

Pre-fetch relationships with selectinload (or joinedload) and the query count stays constant regardless of the row count. It's not that "marshmallow is slow" — it's that the lazy load triggered at dump time is slow — this distinction is decisive for protecting production performance. For the details of SQLAlchemy's loading strategies, see SQLAlchemy 2.0 Practical Guide.

Why is this superior? You can declaratively separate the relationship's "read representation (nesting granularity)" and "write means (foreign key)" within the same group of schemas. Just change only=("id", "title") to adjust the embedding thickness, and even when you add a relationship the view function doesn't change. The relationship's representation is localized to the schema — the practice of ETC.


9. Pagination: A Consistent Response Shape with Envelope and meta

A list API will inevitably need paging as the count grows. Using flask-sqlalchemy's paginate(), assemble a consistent response shape that splits the data into data and the page info into meta.

@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 makes it return an empty array instead of throwing 404 even for a nonexistent page number (safe even if the client pages past the end). Setting an upper bound on per_page (100 here) is a denial-of-service (DoS) protection against an oversized request like per_page=100000 exhausting the DB and memory. The principle of validating external input at the boundary applies equally to paging parameters.

The response is unified to the following shape.

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

Why is this superior? Because every list API returns the same envelope {"data": [...], "meta": {...}}, the client can reuse one kind of parse processing. Having total / pages in the response lets the client assemble a pager UI. Consistency of response shape is the very usability of the API.


10. Testing: Pin the Boundary's Contract with pytest

The boundary's behavior — "valid input is 201, invalid input is 422, a nonexistent resource is 404" — is the API's contract. Only by pinning the contract in tests can you guarantee it won't break under refactoring or feature additions.

Fixtures: Prepare a Test App and DB

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

Round-Trip Verification of Endpoints: 201 / 422 / 404

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

Schema Unit Tests: Verify the Boundary Directly, Without Going Through the View

Tests that hit the schema's load() / dump() directly, without going through an endpoint, are also valuable. As unit tests of the validation logic they're fast and make isolating the cause easy.

# 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

Why is this superior? The security contract — "invalid input is 422, secrets are not output, dump_only can't be written" — is pinned in tests. Even if you refactor the schema in the future, a regression like accidentally removing dump_only is detected instantly by the tests. The boundary's safety is guarded doubly, by declaration (the schema) and by tests (pinning the contract).


Conclusion: Design the Entrance and Exit Together, with a Single Schema

In a Flask + SQLAlchemy REST API, marshmallow-sqlalchemy realizes the "gatekeeper of the boundary" with minimal boilerplate. Here are the key points of this article, restated.

  1. With SQLAlchemyAutoSchema + auto_field(), auto-generate a schema from a SQLAlchemy model and eliminate the double maintenance of model and schema.
  2. With load_instance = True, load() returns a validated ORM instance, and passing session= lets you add() it directly.
  3. With dump_only (mass-assignment prevention) / load_only (secret-leak prevention) / unknown=RAISE (unknown-key rejection), seal the two big security accidents in the structure of the schema.
  4. With @app.errorhandler(ValidationError), convert err.messages to 422 in bulk and eliminate hand-written try / except from each route (DRY).
  5. With fields.Nested / include_relationships, express relationships, and avoid N+1 with selectinload before dumping.
  6. With @post_dump envelope + meta, unify the response shape, and round-trip-verify 201 / 422 / 404 with pytest to pin the boundary's contract.

What separates a "working REST API" from a "REST API you can run for 10 years" is neither the framework choice nor the ORM's speed, but how declaratively and testably you guard the two boundaries of entrance and exit. marshmallow-sqlalchemy is a proven tool for expressing that boundary as a single schema.

For further exploration, I recommend re-reading the following from the official docs with this article's design perspective in mind.


Consult Me on Type-Safe Backend Design

I've implemented and operated the boundary design explained here — "always validate external input at the entrance of a REST API, and shape safely at the exit by dropping secrets" — in the production environment of an economic-ministry-award-winning B2B SaaS, as a Flask / SQLAlchemy / marshmallow stack. From mass-assignment protection, response shaping, ORM integration, and N+1-free relationship design to contract-pinning tests, I build API foundations directly tied to business trustworthiness, fast and high-quality, leveraging generative AI. For REST API development in Python or hardening an existing API, feel free to reach out.

友田

友田 陽大

Developer of a METI Minister's Award–winning product. With TypeScript + Python + AWS, I deliver SaaS, industry DX, and production-grade generative AI (RAG) end to end — from requirements to infrastructure and operations — single-handedly.

Got a challenge?

From design to implementation and operations — solo × generative AI

Implementation like this article's, end to end from requirements to production. Start with a free 30-minute technical consult and tell me about your situation.

Available for both project-based (contract) and advisory engagements. Start with a free 30-minute consult.

Also worth reading