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

> Flask SQLAlchemy Migrate を本番品質で設計・運用する実践ガイド。Flask-SQLAlchemy 3.1とSQLAlchemy 2.0の型付きMapped/mapped_column、session.execute(select)の2.0クエリ、get_or_404・db.paginate、アプリケーションコンテキストと per-request セッション、SQLALCHEMY_ENGINE_OPTIONSのpool_pre_ping/pool_recycleプール設計、SQLALCHEMY_BINDSの読み取りレプリカ、Flask-Migrate（Alembic）のautogenerateレビュー規律とCI/CDでのupgradeまでを公式ドキュメントに忠実な実コードで解説します。

- 公開日: 2026-06-26
- 著者: 友田 陽大
- タグ: Python, Flask, SQLAlchemy, Flask-Migrate, データベース, PostgreSQL, 本番運用, バックエンド
- URL: https://tomodahinata.com/blog/flask-sqlalchemy-flask-migrate-database-production-guide

## 要点

- Flask-SQLAlchemy 3.1はSQLAlchemy 2.0のセッション寿命をFlaskのリクエスト/コンテキストに束ねる拡張。db = SQLAlchemy(model_class=Base)で生成し、init_appで束縛、create_allはwith app.app_context()内でのみ実行する
- モデルは2.0の型付きスタイル（Mapped/mapped_column/relationship）で書く。クエリはModel.queryではなくdb.session.execute(db.select(...))が正道。get_or_404・one_or_404・db.paginateで定型を短くする
- db.sessionはアプリケーションコンテキストにスコープされ、リクエスト/CLI終了時にFlask-SQLAlchemyが自動でteardownする。コンテキストはinit・create_all・直接DBテストなど『リクエスト外』でだけ手で push する
- 本番はSQLALCHEMY_ENGINE_OPTIONSでpool_pre_ping/pool_recycleを設定して死んだ接続を弾く。PgBouncerやサーバレスの背後では必須。TRACK_MODIFICATIONSはオフ、読み取りレプリカはSQLALCHEMY_BINDSで分離する
- 本番DBはcreate_allではなくFlask-Migrate（Alembic）で進化させる。autogenerateは必ずレビュー（リネーム・制約名は検出されない）し、デプロイのflask db upgradeをCI/CDに組み込む。経済産業大臣賞B2B SaaSの運用知見を根拠に解説

---

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

[Flask 本番運用ガイド](/blog/flask-production-guide) で繰り返した通り、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 本番運用ガイド](/blog/flask-production-guide) のデータ層スポーク——です。

筆者は、**経済産業大臣賞を受賞した 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 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide)、無停止マイグレーションの戦略は [Alembic 無停止マイグレーションガイド](/blog/alembic-zero-downtime-migrations-sqlalchemy) に分けています。本記事は「Flask の文脈で」この 2 つをどう束ねるかに集中します。

---

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

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

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

| 関心事 | 生の SQLAlchemy（自前） | Flask-SQLAlchemy（肩代わり） |
|---|---|---|
| エンジン生成 | `create_engine(url, **opts)` を手で呼ぶ | `SQLALCHEMY_DATABASE_URI` / `SQLALCHEMY_ENGINE_OPTIONS` 設定から自動生成 |
| セッション工場 | `sessionmaker` + `scoped_session` を組む | `db.session`（コンテキストにスコープ済み）を提供 |
| リクエスト後の後始末 | `teardown` で `session.remove()` を自分で呼ぶ | コンテキスト終了時に自動でクリーンアップ |
| 複数 DB | 複数エンジン/セッションを手で束ねる | `SQLALCHEMY_BINDS` で宣言的に |
| Flask 統合ヘルパ | なし | `get_or_404` / `paginate` / `db.Model` などを提供 |

公式ドキュメントの言葉を借りれば、**「アクティブな 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 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide) を併読してください。

---

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

[大規模アプリ構成ガイド](/blog/flask-application-factory-blueprints-large-app-structure-guide) で確立した **`extensions.py` の裸の拡張 → `init_app` で束縛**パターンに、Flask-SQLAlchemy を正しく載せます。

### 2.1 `DeclarativeBase` を渡す現行の正準形

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

```python
# 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` で束縛する

```python
# 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 本番運用ガイド](/blog/flask-production-guide) §4 の `from_prefixed_env` 参照）。`migrate.init_app(app, db)` は **`db` を第 2 引数に渡す**点が Flask-SQLAlchemy 以外の拡張と違うので注意してください。

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

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

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

これを忘れると `RuntimeError: Working outside of application context` で落ちます（コンテキストの仕組みは [コンテキスト徹底解説](/blog/flask-application-request-context-g-current-app-guide) と §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()` を使う形で、現行の公式ドキュメントはこのスタイルだけを教えています。

```python
# 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 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide) に深掘りを譲ります。本記事ではリレーションを `relationship` / `back_populates` で双方向に張る最小例だけ示します。

```python
# 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`

```python
# 全ユーザーを 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()` を経由します。

| やりたいこと | 2.0 スタイル |
|---|---|
| 複数件（イテレータ） | `db.session.execute(db.select(User)).scalars()` |
| 複数件（リスト） | `db.session.execute(db.select(User)).scalars().all()` |
| 1 件（無ければ `None`） | `db.session.execute(db.select(User).filter_by(id=1)).scalar_one_or_none()` |
| 1 件（無ければ例外） | `db.session.execute(db.select(User).filter_by(id=1)).scalar_one()` |
| 主キーで取得 | `db.session.get(User, id)`（`select` 不要） |

### 4.2 Flask-SQLAlchemy の `*_or_404` ヘルパ

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

```python
# 主キーで取得、無ければ 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 エラーハンドラ（または [エラー処理・可観測性ガイド](/blog/flask-error-handling-logging-observability-guide) のハンドラ）にそのまま乗ります。ビューに `if user is None: abort(404)` を散らかさずに済みます。

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

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

```python
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 本番運用ガイド](/blog/flask-production-guide) §5 で示した「`get_db` + `teardown_appcontext` を自分で書く」パターンを、DB に関しては拡張が肩代わりしてくれる、ということです。

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

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

```python
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 を触るとき**だけです。

| 場面 | コンテキストは？ | あなたの仕事 |
|---|---|---|
| ビュー関数の中 | リクエストコンテキストが自動で active | 何もしない（`db.session` がそのまま使える） |
| `@app.cli.command` の中 | アプリコンテキストが自動で active | 何もしない |
| 初期化スクリプト・`create_all` | active でない | `with app.app_context():` で push |
| スケジューラ/ワーカーの起動時処理 | active でない | `with app.app_context():` で push |
| DB を直接叩くテスト | テスト次第 | 必要な箇所だけ push |

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

---

## **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(...)` に**そのまま渡される辞書**です。プール関連の主なキーはこれです。

| キー | 役割 | 本番での意味 |
|---|---|---|
| `pool_size` | プールが保持する常設接続数 | ワーカー/同時実行数に合わせる |
| `pool_recycle` | この秒数を超えた接続を再接続 | DB/プロキシ側のアイドルタイムアウトより**短く** |
| `pool_pre_ping` | チェックアウト前に軽い ping で生存確認 | 死んだ接続を掴む事故を防ぐ |
| `pool_timeout` | プール枯渇時に接続を待つ秒数 | 待ちすぎてリクエストを溜めない上限 |

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

```python
# 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 コネクションプーリングガイド](/blog/postgresql-connection-pooling-pgbouncer-serverless-guide) で深掘りしています。Flask-SQLAlchemy 側は「`ENGINE_OPTIONS` に何を渡すか」を決めるだけで、実際のプール挙動は SQLAlchemy/プロキシの領域です。

### 6.2 `SQLALCHEMY_TRACK_MODIFICATIONS` は必ずオフ

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

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

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

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

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

```python
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 つです。

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

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

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

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

| コマンド | 役割 | いつ使うか |
|---|---|---|
| `flask db init` | `migrations/` を生成 | プロジェクト初回のみ |
| `flask db migrate -m "..."` | 差分から script を**自動生成** | モデルを変えた開発時 |
| `flask db upgrade` | DB を前進 | **デプロイ時（CI/CD）** |
| `flask db downgrade` | DB を後退 | ロールバック/ローカル検証 |

`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` を流すのが定石です。

```bash
# デプロイパイプライン（または 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 無停止マイグレーションガイド](/blog/alembic-zero-downtime-migrations-sqlalchemy) に深掘りを譲ります。本記事の `flask db upgrade` が「どう流すか」を、その記事が「どんな順序で安全に当てるか」を担います。

---

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

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

```python
# 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 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide)）。
- **外部入力は許可リストで検証**する（`status` の許可値、`items` の型）。`request.args` / `request.get_json()` は信頼できない入力です。本格的なスキーマ検証は marshmallow に寄せます——モデルを API 境界でシリアライズ/デシリアライズする層分離は [marshmallow × Flask × SQLAlchemy で本番 REST API を設計する](/blog/marshmallow-flask-sqlalchemy-rest-api-production-guide) で詳説しています。本記事が「DB をどう叩くか」を、その記事が「境界でどう検証/整形するか」を担います。
- **`db.paginate(stmt, max_per_page=100)`** で `per_page` の上限を縛り、`?per_page=100000` のような DoS 的リクエストを防ぐ。
- **書き込みは `add` → `commit`、例外は `rollback`**（§5.2）。`commit` を必ず通る経路に置き、失敗時はセッションを巻き戻してから 500 を返す。

---

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

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

| # | 項目 | 判定基準 | 本記事 |
|---|---|---|---|
| 1 | `db = SQLAlchemy(model_class=Base)` で生成しているか | `DeclarativeBase` 継承の `Base` を渡している | §2.1 |
| 2 | 拡張は `extensions.py` で裸 → ファクトリで `init_app` か | `db.init_app(app)` / `migrate.init_app(app, db)` | §2.2 |
| 3 | `create_all` を本番スキーマ管理に使っていないか | 本番は Flask-Migrate のみ。`create_all` はテスト/試作だけ | §2.3 / §7 |
| 4 | モデルは型付き（`Mapped` / `mapped_column`）か | 新規モデルは 2.0 スタイル。`db.Column` 新規追加なし | §3 |
| 5 | クエリは `db.session.execute(db.select(...))` か | `Model.query` を新規コードで使っていない | §4 |
| 6 | 書き込みに `commit`、例外に `rollback` があるか | 自動 commit はされない。`rollback` 漏れがない | §5.2 |
| 7 | コンテキスト外の DB アクセスを `app_context()` で囲んだか | 初期化/バッチ/直接テストで push している | §5.3 |
| 8 | `pool_pre_ping` / `pool_recycle` を設定したか | プロキシ/サーバレス背後で死んだ接続を弾く | §6.1 |
| 9 | `SQLALCHEMY_TRACK_MODIFICATIONS=False` か | 既定オフを維持。ORM イベントで代替 | §6.2 |
| 10 | 読み取りレプリカ/複数 DB は `SQLALCHEMY_BINDS` か | bind key で分離している | §6.3 |
| 11 | autogenerate を目視レビューしたか | drop/add 誤検出・リネーム・downgrade を確認 | §7.2 |
| 12 | `flask db upgrade` をデプロイで 1 回だけ流すか | 複数ワーカー/タスクが同時に走らせない | §7.3 |
| 13 | 全クエリを `tenant_id` でスコープしたか（マルチテナント） | テナント越境が構造的に起きない | §8 |
| 14 | N+1 を eager loading で潰したか | 一覧で `selectinload` 等を当てている | §8 |
| 15 | API 境界で外部入力を検証したか | 許可リスト + marshmallow で境界を固める | §8 |

---

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

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 本番運用ガイド](/blog/flask-production-guide) に戻って俯瞰し、ORM のモデリングは [SQLAlchemy 2.0 実践ガイド](/blog/sqlalchemy-2-typed-orm-production-guide)、無停止マイグレーションは [Alembic 無停止マイグレーションガイド](/blog/alembic-zero-downtime-migrations-sqlalchemy) へ——Flask のデータ層は、この 3 本で「載せ方・モデリング・進化」が揃います。