Flaskのデータ層：Flask-SQLAlchemy 3.x（2.0スタイル）とFlask-Migrateで本番DBを設計・運用する

導入：Flask は「データ層」を持たない

Flask 本番運用ガイド で繰り返した通り、Flask は「核だけ」を提供するフレームワークです。その「持たないもの」の筆頭が データ層——ORM もコネクションプールもマイグレーションも、Flask 本体には一切ありません。だからこそ本番 Flask では、「データ層をどう載せるか」が設計判断の中心になります。

その答えの定番が Flask-SQLAlchemy と Flask-Migrate です。前者は SQLAlchemy（Python の事実上標準 ORM）を Flask のリクエスト/コンテキストのライフサイクルに束ね、後者は Alembic（SQLAlchemy 公式のマイグレーションツール）を Flask の flask db CLI として使えるようにします。本記事は、この 2 つを Flask-SQLAlchemy 3.1 / SQLAlchemy 2.0 / Flask-Migrate 4.1 の公式ドキュメントに忠実に、本番品質で組み上げる深掘りガイド——Flask 本番運用ガイド のデータ層スポーク——です。

筆者は、経済産業大臣賞を受賞した B2B SaaS のバックエンドを Python / Flask / SQLAlchemy / PostgreSQL で設計・実装し、ALB → ECS(Fargate) 上で本番運用してきました。マルチテナントの請求・在庫・取引データを扱う中で、「セッションをどこで commit/rollback するか」「死んだ接続をどう弾くか」「スキーマをどう無停止で進化させるか」は、机上の知識ではなく毎日の運用課題でした。本記事はその実戦の設計判断を、公式仕様に照らして言語化したものです。

💡 この記事で扱うバージョン：Flask-SQLAlchemy 3.1.1（SQLAlchemy >= 2.0.16、Python >= 3.8 が必要）と Flask-Migrate 4.1.0（Alembic をラップ）を前提とします。SQLAlchemy 2.0 系の型付き ORM とクエリスタイルを採用します。ORM のモデリング自体の深掘りは SQLAlchemy 2.0 実践ガイド、無停止マイグレーションの戦略は Alembic 無停止マイグレーションガイド に分けています。本記事は「Flask の文脈で」この 2 つをどう束ねるかに集中します。

1. なぜ Flask-SQLAlchemy か：生の SQLAlchemy との違い

最初に答えるべき問いは「生の SQLAlchemy を直接使えばよいのでは？」です。SQLAlchemy 単体でも完結したライブラリですし、FastAPI などでは生の SQLAlchemy を使うのが一般的です。それでも Flask では Flask-SQLAlchemy を使う理由があります。

Flask-SQLAlchemy が肩代わりするのは、「SQLAlchemy のセッション寿命を Flask のリクエスト/コンテキストに束ねる」配線です。生の SQLAlchemy でこれを自前で書くと、次の定型コードを全部自分で管理することになります。

公式ドキュメントの言葉を借りれば、「アクティブな Flask アプリケーションコンテキストが、クエリの発行と db.engine / db.session へのアクセスに必須」 であり、「セッションはコンテキストにスコープされるので、各リクエストや CLI コマンドの後で適切にクリーンアップされる」。この「コンテキストへのスコープと自動クリーンアップ」こそが、Flask-SQLAlchemy が肩代わりする価値の中核です（§5 で深掘りします）。

💡 「どちらが優れているか」ではなく「どちらが Flask に馴染むか」：生の SQLAlchemy が劣るわけではありません。Flask の request ライフサイクルに乗るアプリでは、Flask-SQLAlchemy の自動配線が定型を消してくれる——それだけです。逆に「Flask のコンテキスト外で大量のバッチ処理を回す」のが主用途なら、生の SQLAlchemy + 明示的なセッション管理のほうが見通しが良いこともあります。本記事は前者（リクエスト駆動の Web/API）を前提とします。SQLAlchemy 2.0 の Mapped / select() / Session そのものの設計は SQLAlchemy 2.0 実践ガイド を併読してください。

2. セットアップ：ファクトリと db = SQLAlchemy(model_class=Base)

大規模アプリ構成ガイド で確立した extensions.py の裸の拡張 → init_app で束縛パターンに、Flask-SQLAlchemy を正しく載せます。

2.1 DeclarativeBase を渡す現行の正準形

Flask-SQLAlchemy 3.1 の公式が推奨する生成方法は、SQLAlchemy 2.0 の DeclarativeBase を継承した Base を model_class に渡す形です。これにより db.Model が 2.0 の宣言的ベースとして振る舞い、後述の型付きモデルが自然に書けます。

# src/myapp/extensions.py — どのアプリにも束縛されていない「裸」の拡張
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


db = SQLAlchemy(model_class=Base)   # この時点では app を知らない
migrate = Migrate()                  # この時点では app を知らない

Base を自分で定義しておくと、共通のミックスイン（created_at / updated_at の自動付与など）や型アノテーションのデフォルトをここに集約できます。これは「将来の拡張」ではなく、本番では早晩必ず欲しくなる共通基盤なので、最初から Base を切り出しておく価値があります。

2.2 ファクトリで設定し、init_app で束縛する

# src/myapp/__init__.py（抜粋）
from flask import Flask

from .extensions import db, migrate


def create_app(test_config: dict | None = None) -> Flask:
    app = Flask(__name__, instance_relative_config=True)

    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///project.db"
    if test_config is not None:
        app.config.update(test_config)

    # ここで初めて app に束縛する（二段階初期化の第二段）
    db.init_app(app)
    migrate.init_app(app, db)

    from .blueprints.api import bp as api_bp
    app.register_blueprint(api_bp, url_prefix="/api")

    return app

SQLALCHEMY_DATABASE_URI は 文字列または SQLAlchemy の URL オブジェクトを受け取ります。本番では当然この値を環境変数から流し込みます（FLASK_SQLALCHEMY_DATABASE_URI='postgresql+psycopg://...'、Flask 本番運用ガイド §4 の from_prefixed_env 参照）。migrate.init_app(app, db) は db を第 2 引数に渡す点が Flask-SQLAlchemy 以外の拡張と違うので注意してください。

2.3 create_all は「アプリケーションコンテキスト内」でだけ

テーブルを物理的に作る db.create_all() は、アクティブなアプリケーションコンテキストを要求します。ファクトリで作ったアプリにはリクエスト外では自動のコンテキストがないため、手で push する必要があります。

with app.app_context():
    db.create_all()

これを忘れると RuntimeError: Working outside of application context で落ちます（コンテキストの仕組みは コンテキスト徹底解説 と §5 を参照）。

⚠️ create_all は本番のスキーマ管理に使わない：db.create_all() は「まだ存在しないテーブルを作るだけ」です。既存テーブルへのカラム追加・型変更・制約追加・削除は一切行いません。つまり一度作ったテーブルを進化させる手段がありません。create_all が許されるのは、テストの使い捨て DB や最初のローカル試作までです。本番のスキーマ管理は §7 の Flask-Migrate（Alembic）が唯一の正解であり、create_all を本番デプロイのスクリプトに書くのは、スキーマ変更を二度とできない罠への片道切符です。

3. モデル定義：SQLAlchemy 2.0 の型付きスタイル

Flask-SQLAlchemy 3.1 + SQLAlchemy 2.0 では、モデルを 型付き（typed）スタイルで書きます。Mapped[...] 型アノテーションと mapped_column() を使う形で、現行の公式ドキュメントはこのスタイルだけを教えています。

# src/myapp/models/user.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from myapp.extensions import db


class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)
    email: Mapped[str]

ここで起きていることを正確に読み解きます。

• Mapped[int] / Mapped[str]：カラムの Python 型を型アノテーションで宣言します。型チェッカー（mypy / Pyright）が user.email を str と理解できるようになり、db.Column("email", String) 時代の「型が一切伝わらない」問題が消えます。
• mapped_column(...)：カラムの DB 側の制約（primary_key / unique / nullable など）を指定します。Mapped[str]（Optional でない）は NOT NULL に、Mapped[str | None] は NULL 可にマッピングされます——型が DB 制約を駆動するのが 2.0 の肝です。
• email: Mapped[str] のように mapped_column() を省くと、型アノテーションだけからカラムが推論されます（追加の制約が要らない単純カラムはこれで足ります）。

💡 db.Column は「非推奨」ではなく「旧スタイルのエイリアス」：誤解されがちですが、db.Column が削除されたわけでも、非推奨警告が出るわけでもありません。SQLAlchemy 2.0 では db.Column は今もレガシーなエイリアスとして存在します。ただし現行の公式ドキュメントは型付きの Mapped / mapped_column スタイルだけを教えており、新規コードはこちらで書くのが本筋です。既存の db.Column モデルを今すぐ全部書き換える必要はありませんが、新しく書くモデルは型付きスタイルで揃える——これが負債を増やさない方針です。

リレーション・複合制約・インデックス・ハイブリッドプロパティといった本格的なモデリングは、Flask 固有ではなく SQLAlchemy 2.0 の領域なので、SQLAlchemy 2.0 実践ガイド に深掘りを譲ります。本記事ではリレーションを relationship / back_populates で双方向に張る最小例だけ示します。

# src/myapp/models/order.py
from datetime import datetime

from sqlalchemy import ForeignKey, Index, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from myapp.extensions import db


class Order(db.Model):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    # マルチテナント B2B SaaS：全行に tenant_id を持たせて分離する（§8）
    tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
    status: Mapped[str] = mapped_column(String(32), default="pending")
    total: Mapped[int]
    created_at: Mapped[datetime] = mapped_column(default=func.now())

    items: Mapped[list["OrderItem"]] = relationship(back_populates="order")

    # 複合インデックス：テナント単位の一覧クエリ（§8）を効かせる
    __table_args__ = (
        Index("ix_orders_tenant_created", "tenant_id", "created_at"),
    )


class OrderItem(db.Model):
    __tablename__ = "order_items"

    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), index=True)
    product_name: Mapped[str]
    quantity: Mapped[int]

    order: Mapped["Order"] = relationship(back_populates="items")

tenant_id に index=True、(tenant_id, created_at) に複合インデックスを張っている点が、§8 のマルチテナント一覧クエリと N+1 対策の伏線です。

4. クエリ：2.0 スタイル（session.execute(select(...))）が正道

ここが SQLAlchemy 1.x からの最大の変化点です。Model.query / session.query() は SQLAlchemy では「レガシー（legacy in SQLAlchemy）」 と公式に明言されました。現行の正道は db.session.execute(db.select(...)) です。

4.1 基本：select → execute → scalars

# 全ユーザーを username 順で取得
users = db.session.execute(
    db.select(User).order_by(User.username)
).scalars()

db.select(User) でクエリを組み、db.session.execute(...) で実行し、.scalars() で 「行（Row）」ではなく「エンティティ（User）」の列を取り出します。.scalars() を付けないと (User,) のようなタプルの行が返るので、ORM エンティティが欲しいときは必ず .scalars() を経由します。

4.2 Flask-SQLAlchemy の *_or_404 ヘルパ

Web の文脈では「見つからなければ 404」が頻出します。Flask-SQLAlchemy はこれを 1 行にする統合ヘルパを提供します。

# 主キーで取得、無ければ 404 を送出
user = db.get_or_404(User, id)

# select の結果が 1 件、無ければ 404（description でメッセージを差し替え可能）
user = db.one_or_404(
    db.select(User).filter_by(username=username),
    description=f"User {username} not found.",
)

# select の最初の 1 件、無ければ 404
user = db.first_or_404(db.select(User).order_by(User.created_at))

これらは内部で abort(404) を呼ぶので、§6 で設定する JSON エラーハンドラ（または エラー処理・可観測性ガイド のハンドラ）にそのまま乗ります。ビューに if user is None: abort(404) を散らかさずに済みます。

4.3 ページネーション：db.paginate

一覧 API には db.paginate を使います。select を渡すと Pagination オブジェクトが返り、ページ番号・件数・前後ページの有無を持ちます。

page = db.paginate(db.select(User).order_by(User.join_date))

page.items        # このページの要素（list）
page.page         # 現在のページ番号
page.total        # 全件数
page.has_next     # 次ページの有無
page.has_prev     # 前ページの有無
page.iter_pages() # ページャ UI 用のページ番号イテレータ

page / per_page は既定でリクエストのクエリ文字列（?page=2&per_page=20）から読まれます。これにより「クエリ文字列を手でパースしてオフセットを計算する」定型が消えます。実際の一覧エンドポイントへの組み込みは §8 で示します。

⚠️ Model.query がまだ動くからといって新規コードで使わない：Flask-SQLAlchemy は後方互換のため User.query を今も提供しますが、その実体は SQLAlchemy 側でレガシー扱いの Query インターフェースです。User.query.filter_by(...).first() のような書き方は動きはしますが、2.0 の select ベースと比べて型推論が弱く、async セッションへの移行口もなく、SQLAlchemy 公式の将来方針から外れています。新規コードは db.session.execute(db.select(...)) で統一し、Model.query は既存コードを触るとき以外は使わない——これが 2.0 時代の規律です。

5. アプリケーションコンテキストと per-request セッション

Flask-SQLAlchemy の挙動を「魔法」で済ませず、コンテキストとセッションの関係を正確に理解します。ここが本番で DetachedInstanceError や「commit したのに反映されない」といった事故を防ぐ分岐点です。

5.1 公式が言う「コンテキストにスコープされたセッション」

公式の最重要事実を 2 つ、そのまま引きます。

1. 「アクティブな Flask アプリケーションコンテキストが、クエリの発行と db.engine / db.session へのアクセスに必須」
2. 「セッションはコンテキストにスコープされるので、各リクエストや CLI コマンドの後で適切にクリーンアップされる」

つまり db.session は、いま処理中のアプリケーションコンテキスト（≒ 1 リクエスト、または 1 CLI コマンド）に紐づいたスコープ付きセッションです。リクエストごとに独立したセッションが割り当てられ、リクエストが終わると Flask-SQLAlchemy が自動でそのセッションを後始末（remove）してくれます。あなたが teardown を書く必要はありません——これが Flask 本番運用ガイド §5 で示した「get_db + teardown_appcontext を自分で書く」パターンを、DB に関しては拡張が肩代わりしてくれる、ということです。

5.2 commit / rollback の責任はあなたにある

自動なのは**セッションの寿命管理（割り当てと後始末）**であって、トランザクションの確定（commit）ではありません。書き込みは必ず明示的に commit し、失敗時は rollback します。

from sqlalchemy.exc import SQLAlchemyError


def create_user(username: str, email: str) -> User:
    user = User(username=username, email=email)
    db.session.add(user)
    try:
        db.session.commit()
    except SQLAlchemyError:
        db.session.rollback()   # 失敗したら必ず巻き戻す
        raise
    return user


def delete_user(user: User) -> None:
    db.session.delete(user)
    db.session.commit()

db.session.add(obj) / db.session.delete(obj) で変更をステージし、db.session.commit() で確定します。例外時に rollback を怠ると、セッションが「壊れた（失敗トランザクションが残った）」状態になり、後続のクエリが PendingRollbackError で連鎖失敗します。

💡 リクエスト終了時、Flask-SQLAlchemy は自動 commit しない：勘違いしやすい点です。拡張がやるのは session.remove()（セッションの後始末）であって、commit ではありません。commit していない変更はリクエスト終了とともに破棄されます。「add したのに DB に入っていない」事故のほぼ全ては、commit の呼び忘れです。書き込み系のビューでは commit を必ず通る経路に置いてください。

5.3 コンテキストを「手で push する」のはいつか

リクエスト処理中や flask CLI コマンド実行中は、Flask が自動でコンテキストを push してくれるので、あなたが意識する必要はありません。手で with app.app_context(): を書くのは、リクエスト/CLI の外側で DB を触るときだけです。

公式のテストに関する指針は明確です——「テストごとに、コンテキストを必要な場所で・必要な時間だけ push せよ（Only push a context exactly where and for how long it's needed for each test）」。テスト全体を巨大な app_context() で包むのではなく、DB アクセスが必要なブロックだけをピンポイントで包む、ということです。テストの組み立て方は Flask 本番運用ガイド §9 と各テスト記事に譲ります。コンテキストの push/pop の機構そのものは コンテキスト徹底解説 を参照してください。

6. 本番のコネクションプールと設定

開発では SQLite で動いても、本番（PostgreSQL on RDS/Aurora、しばしば PgBouncer 越し）ではコネクションプールの設定が安定性を左右します。Flask-SQLAlchemy はこれを SQLALCHEMY_ENGINE_OPTIONS 経由で制御します。

6.1 SQLALCHEMY_ENGINE_OPTIONS：create_engine への直結

SQLALCHEMY_ENGINE_OPTIONS は、Flask-SQLAlchemy が内部で呼ぶ SQLAlchemy の create_engine(...) にそのまま渡される辞書です。プール関連の主なキーはこれです。

筆者が本番で使う最小かつ効く設定がこれです（推奨例であり、公式の逐語引用ではありません——適切な値はワークロードと DB 側設定に依存します）。

# create_app 内、または ProductionConfig
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
    "pool_pre_ping": True,   # 死んだ接続をチェックアウト時に弾く
    "pool_recycle": 300,     # 5 分でリサイクル（プロキシのアイドル切断対策）
}

pool_pre_ping と pool_recycle は、ロードバランサや PgBouncer、サーバレス DB がアイドル接続を黙って切る環境で決定的に重要です。これらを設定しないと、プールに残った「相手側ではもう死んでいる接続」を掴んで OperationalError: server closed the connection unexpectedly を踏みます。pool_pre_ping=True はチェックアウト時に軽量な ping を打ってこれを未然に弾き、pool_recycle は古い接続を能動的に作り直します。

💡 PgBouncer / サーバレスの背後では SQLAlchemy 側のプールを「薄く」する：PgBouncer（transaction モード）や RDS Proxy、サーバレス Postgres を挟むと、接続プーリングはプロキシ側が担うため、SQLAlchemy 側で厚いプールを持つと二重プールになって接続が溢れます。この場合は pool_size を小さく、もしくは NullPool を検討し、pool_pre_ping は引き続き有効にします。プロキシ前提のプール設計（prepared statement の扱い、transaction/session モードの違いを含む）は PostgreSQL コネクションプーリングガイド で深掘りしています。Flask-SQLAlchemy 側は「ENGINE_OPTIONS に何を渡すか」を決めるだけで、実際のプール挙動は SQLAlchemy/プロキシの領域です。

6.2 SQLALCHEMY_TRACK_MODIFICATIONS は必ずオフ

SQLALCHEMY_TRACK_MODIFICATIONS は Flask-SQLAlchemy 3.0 以降、既定で無効です。これは意図的な既定であり、有効化してはいけません。公式の言葉では——「これは各セッションに無視できないオーバーヘッドを追加する。SQLAlchemy の ORM イベントを直接使うことを推奨する」。

# 既定でオフだが、Config で意図を明示しておくと安全
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

オブジェクト変更の通知が本当に必要なら、この機能ではなく SQLAlchemy のネイティブな ORM イベント（before_flush など）を使います。

6.3 SQLALCHEMY_BINDS：複数 DB と読み取りレプリカ

複数のデータベースを 1 アプリで扱うときは SQLALCHEMY_BINDS を使います。キー（bind key）ごとに別の URI を宣言し、モデルやクエリで bind を指定します。

app.config["SQLALCHEMY_DATABASE_URI"] = "postgresql+psycopg://primary/app"
app.config["SQLALCHEMY_BINDS"] = {
    # 読み取りレプリカ：重い集計・一覧読み取りをプライマリから逃がす
    "read_replica": "postgresql+psycopg://replica/app",
    # 別系統 DB（例：分析用）
    "analytics": "postgresql+psycopg://analytics/warehouse",
}

モデル単位で bind を固定するなら __bind_key__ を、クエリ単位で振り分けるなら実行時に bind を指定します。読み取りレプリカへのルーティングは「読み取り専用クエリだけをレプリカに流し、書き込みとレプリカ遅延に敏感な読み取りはプライマリに残す」という設計判断が要るため、SQLALCHEMY_BINDS は配線の入口であって、ルーティング戦略そのものは別途設計が必要です。

7. Flask-Migrate：本番スキーマを Alembic で進化させる

create_all（§2.3）が使えるのは試作までで、本番のスキーマ進化は Flask-Migrate（Alembic）が唯一の手段です。公式の定義はこうです——「Flask-Migrate は Alembic を Flask と Flask-SQLAlchemy アプリケーションで適切に動くように設定する拡張で……すべては Alembic が処理する」。Flask-Migrate 自体は薄いラッパで、マイグレーションの実体は Alembic です。

7.1 セットアップと 4 つの CLI

migrate.init_app(app, db)（§2.2）を済ませてあれば、flask db サブコマンドが使えます。本番運用で使うのは実質この 4 つです。

# ① マイグレーション環境を初期化（最初の一度だけ）。migrations/ を生成
flask db init

# ② モデルの変更を検出して、マイグレーションスクリプトを生成
flask db migrate -m "add tenant_id to orders"

# ③ マイグレーションを適用（DB を最新スキーマへ前進）
flask db upgrade

# ④ 直前のマイグレーションを巻き戻す
flask db downgrade

migrate -m "..." で生成された migrations/versions/*.py は、必ず Git にコミットします。これがスキーマ変更の唯一の真実（single source of truth）になり、全環境（開発・ステージング・本番）で同じ順序のマイグレーションが適用される保証になります。

7.2 autogenerate は「下書き」——必ずレビューする

ここが本番で最も事故るポイントです。flask db migrate の自動生成（autogenerate）は便利ですが、完璧ではありません。公式の警告をそのまま引きます——「マイグレーションスクリプトはレビューと編集が必要だ。Alembic は常にすべての変更を検出できるわけではない……Alembic は現状、テーブル名の変更・カラム名の変更・匿名で命名された制約を検出できない」。

具体的に何が起きるか。

• カラムの「リネーム」を「削除 + 追加」と誤検出する：old_name → new_name のリネームを、Alembic は drop_column("old_name") + add_column("new_name") として生成します。これをそのまま流すとそのカラムのデータが全消失します。リネームは op.alter_column(..., new_column_name=...) に手で書き換える必要があります。
• テーブル名の変更を検出しない：同様に「削除 + 新規作成」になり、データが消えます。
• 匿名命名の制約を検出しない：名前のない制約（UniqueConstraint(...) を無名で付けたものなど）の変更は追従できません。

⚠️ autogenerate を無レビューでコミット・適用しない（本番事故の最頻原因）：flask db migrate の出力を git add する前に、生成された upgrade() / downgrade() を必ず目視レビューしてください。特に「drop_column が出ていないか」「リネームのつもりが drop+add になっていないか」「downgrade() が upgrade() を正しく逆転しているか」の 3 点は毎回確認します。筆者の運用では、autogenerate されたマイグレーションのレビューを通常のコードレビューと同じ PR の必須項目にしていました。スキーマ変更は一度本番に流れると巻き戻しが高くつくため、ここのレビュー規律が本番 DB の生死を分けます。

7.3 CI/CD とコンテナでマイグレーションを流す

本番では、デプロイのたびに flask db upgrade を自動実行してスキーマを前進させます。ECS(Fargate) のようなコンテナ環境では、アプリ起動の「前」に、専用の単発タスクとして flask db upgrade を流すのが定石です。

# デプロイパイプライン（または ECS の事前タスク）で実行
flask db upgrade
# 成功したらアプリ（Gunicorn）を起動
gunicorn -w 4 'myapp:create_app()'

⚠️ 複数ワーカー/タスクが同時に upgrade を走らせない：Gunicorn の各ワーカーやスケールアウトした複数の Fargate タスクが、それぞれ起動時に flask db upgrade を呼ぶ設計にすると、同じマイグレーションが競合して二重適用やロックで詰まります。マイグレーションはデプロイの 1 ステップ（単一の事前タスク/ジョブ）として 1 回だけ実行し、アプリのワーカー群はマイグレーション済みの DB に対して起動するだけにしてください。

スキーマ変更をダウンタイムなしで当てる（既存のアプリと新スキーマを共存させる）には、upgrade を当てる順序とコードのデプロイ順序を噛み合わせる expand-and-contract（拡張と収縮） パターンが必要です。「カラムを追加 → 新旧コードが両方読める状態を作る → 旧カラムを後で削除」という多段デプロイの設計は、Flask の領域を超えるので Alembic 無停止マイグレーションガイド に深掘りを譲ります。本記事の flask db upgrade が「どう流すか」を、その記事が「どんな順序で安全に当てるか」を担います。

8. 本番の実例：マルチテナント一覧 + 作成エンドポイント

ここまでの要素を、マルチテナント B2B SaaS の現実的なエンドポイントに統合します。テナント分離・ページネーション・フィルタ・N+1 対策・commit/rollback を 1 つの Blueprint に落とします。

# src/myapp/blueprints/api/orders.py
from flask import Blueprint, abort, jsonify, request
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import selectinload

from myapp.extensions import db
from myapp.models.order import Order, OrderItem

bp = Blueprint("orders", __name__)


def current_tenant_id() -> int:
    """認証ミドルウェアが検証・格納したテナントIDを取り出す。
    全クエリをこの tenant_id で必ずスコープするのがマルチテナント分離の生命線。"""
    tenant_id = getattr(request, "tenant_id", None)
    if tenant_id is None:
        abort(401, description="tenant not resolved")
    return tenant_id


@bp.get("/orders")
def list_orders():
    tenant_id = current_tenant_id()

    # ① select を tenant_id で必ずスコープ（テナント越境を構造的に防ぐ）
    stmt = db.select(Order).where(Order.tenant_id == tenant_id)

    # ② 任意フィルタ（外部入力なので許可値だけを通す）
    status = request.args.get("status")
    if status is not None:
        if status not in {"pending", "paid", "shipped", "cancelled"}:
            abort(400, description="invalid status filter")
        stmt = stmt.where(Order.status == status)

    # ③ N+1 を防ぐ：items を selectinload で一括ロード（§8 N+1 awareness）
    stmt = stmt.options(selectinload(Order.items)).order_by(Order.created_at.desc())

    # ④ db.paginate がクエリ文字列(?page=&per_page=)を読んでページング
    page = db.paginate(stmt, max_per_page=100)

    return jsonify(
        items=[
            {
                "id": o.id,
                "status": o.status,
                "total": o.total,
                "item_count": len(o.items),   # selectinload 済みなので追加クエリ無し
            }
            for o in page.items
        ],
        page=page.page,
        total=page.total,
        has_next=page.has_next,
    )


@bp.post("/orders")
def create_order():
    tenant_id = current_tenant_id()
    payload = request.get_json(silent=True) or {}

    # 境界バリデーション（ここでは最小限。実務は marshmallow へ）
    items = payload.get("items")
    if not isinstance(items, list) or not items:
        abort(400, description="items must be a non-empty list")

    order = Order(tenant_id=tenant_id, status="pending", total=0)
    total = 0
    for raw in items:
        qty = raw.get("quantity", 0)
        order.items.append(
            OrderItem(product_name=str(raw.get("product_name", "")), quantity=qty)
        )
        total += qty
    order.total = total

    db.session.add(order)
    try:
        db.session.commit()        # ここで初めて確定（§5.2）
    except SQLAlchemyError:
        db.session.rollback()       # 失敗したら必ず巻き戻す
        abort(500, description="failed to create order")

    return jsonify(id=order.id, total=order.total), 201

この実装に込めた本番の規律を整理します。

• テナント分離は「全 select を tenant_id でスコープ」で構造的に担保する。where(Order.tenant_id == tenant_id) を入口で必ず当てる。これを忘れると他テナントのデータが見えるという、SaaS で最も重い事故になります。アプリ層だけに頼らず、PostgreSQL の RLS で DB 層でも二重に縛るのが堅牢ですが、まずはアプリ層で全クエリをスコープするのが最低ライン。
• N+1 を selectinload で潰す。一覧で各 order.items を遅延ロードすると、行数 N に対して N+1 回のクエリが飛びます。options(selectinload(Order.items)) で items を 1 回の追加クエリでまとめてロードし、len(o.items) がループ内で追加クエリを起こさないようにします（N+1 と eager loading 戦略の詳細は SQLAlchemy 2.0 実践ガイド）。
• 外部入力は許可リストで検証する（status の許可値、items の型）。request.args / request.get_json() は信頼できない入力です。本格的なスキーマ検証は marshmallow に寄せます——モデルを API 境界でシリアライズ/デシリアライズする層分離は marshmallow × Flask × SQLAlchemy で本番 REST API を設計する で詳説しています。本記事が「DB をどう叩くか」を、その記事が「境界でどう検証/整形するか」を担います。
• db.paginate(stmt, max_per_page=100) で per_page の上限を縛り、?per_page=100000 のような DoS 的リクエストを防ぐ。
• 書き込みは add → commit、例外は rollback（§5.2）。commit を必ず通る経路に置き、失敗時はセッションを巻き戻してから 500 を返す。

9. 本番DB設計チェックリスト

これまでの要点を、レビュー前に通す実務チェックリストにまとめます。

まとめ：データ層は「自動の寿命」と「明示の確定」で設計する

Flask 本体はデータ層を持たないからこそ、Flask-SQLAlchemy と Flask-Migrate を正しく束ねる設計が本番品質を決めます。本記事の規律を再掲します。

1. db = SQLAlchemy(model_class=Base) で生成し、extensions.py の裸の拡張 → ファクトリで init_app する。create_all は with app.app_context(): 内でのみ、かつ試作/テスト限定。
2. モデルは SQLAlchemy 2.0 の型付きスタイル（Mapped / mapped_column / relationship）で書く。db.Column はレガシーエイリアスとして残るが、新規は型付きで揃える。
3. クエリは db.session.execute(db.select(...)) が正道。Model.query はレガシー。get_or_404 / one_or_404 / db.paginate で Web の定型を短くする。
4. db.session の寿命はコンテキストにスコープされ自動で後始末されるが、commit / rollback の責任はあなたにある。リクエスト終了時に自動 commit はされない。
5. 本番プールは pool_pre_ping / pool_recycle で死んだ接続を弾く（PgBouncer/サーバレス背後では必須）。TRACK_MODIFICATIONS はオフ、読み取りレプリカは SQLALCHEMY_BINDS。
6. スキーマ進化は Flask-Migrate（Alembic）で。autogenerate は下書きであり、リネーム・制約は検出されないので必ずレビュー。flask db upgrade はデプロイで 1 回だけ流す。
7. マルチテナントは全クエリを tenant_id でスコープし、N+1 は eager loading で潰し、境界は許可リスト + marshmallow で固める。

これらは「規模が大きくなったら」ではなく、最初のテーブルから効く設計です。全体像は Flask 本番運用ガイド に戻って俯瞰し、ORM のモデリングは SQLAlchemy 2.0 実践ガイド、無停止マイグレーションは Alembic 無停止マイグレーションガイド へ——Flask のデータ層は、この 3 本で「載せ方・モデリング・進化」が揃います。