Flask's Data Layer: Designing and Operating a Production DB with Flask-SQLAlchemy 3.x (2.0 Style) and Flask-Migrate

Introduction: Flask Has No "Data Layer"

As repeated in the Flask production-operations guide, Flask is a framework that provides "only the core." The foremost of its "doesn't-haves" is the data layer — there is no ORM, no connection pool, no migration in Flask itself. That's exactly why, in production Flask, "how to load the data layer" becomes the center of design judgment.

The standard answer is Flask-SQLAlchemy and Flask-Migrate. The former binds SQLAlchemy (Python's de facto standard ORM) to Flask's request/context lifecycle, and the latter makes Alembic (SQLAlchemy's official migration tool) usable as Flask's flask db CLI. This article is a deep-dive guide to assembling these 2 at production quality, faithful to the official documentation of Flask-SQLAlchemy 3.1 / SQLAlchemy 2.0 / Flask-Migrate 4.1 — the data-layer spoke of the Flask production-operations guide.

The author designed and implemented the backend of a B2B SaaS that won the Minister of Economy, Trade and Industry Award in Python / Flask / SQLAlchemy / PostgreSQL, and ran it in production on ALB → ECS (Fargate). Handling multi-tenant billing, inventory, and transaction data, "where to commit/rollback the session," "how to reject dead connections," and "how to evolve the schema without downtime" were not desk knowledge but daily operational tasks. This article verbalizes those field design judgments, against the official specs.

💡 The versions covered in this article: it assumes Flask-SQLAlchemy 3.1.1 (requires SQLAlchemy >= 2.0.16, Python >= 3.8) and Flask-Migrate 4.1.0 (which wraps Alembic). It adopts the SQLAlchemy 2.0-series typed ORM and query style. The deep-dive on ORM modeling itself is split into the SQLAlchemy 2.0 practical guide, and the strategy of zero-downtime migration into the Alembic zero-downtime migration guide. This article concentrates on how to bind these 2 "in the Flask context."

1. Why Flask-SQLAlchemy: The Difference from Raw SQLAlchemy

The first question to answer is "couldn't I just use raw SQLAlchemy directly?" SQLAlchemy alone is a self-contained library, and in FastAPI etc. using raw SQLAlchemy is common. Even so, there's a reason to use Flask-SQLAlchemy in Flask.

What Flask-SQLAlchemy takes off your hands is the wiring that "binds SQLAlchemy's session lifetime to Flask's request/context." Write this yourself with raw SQLAlchemy and you'll manage all the following boilerplate yourself.

In the official documentation's words, "an active Flask application context is required to issue queries and to access db.engine / db.session," and "because sessions are scoped to the context, they are cleaned up appropriately after each request or CLI command." This "scoping to the context and auto-cleanup" is the core of the value Flask-SQLAlchemy takes off your hands (deep-dived in §5).

💡 Not "which is superior" but "which fits Flask": raw SQLAlchemy isn't inferior. In an app riding on Flask's request lifecycle, Flask-SQLAlchemy's auto-wiring erases the boilerplate — that's all. Conversely, if the main use is "spinning large batch processing outside the Flask context," raw SQLAlchemy + explicit session management can be clearer. This article presupposes the former (request-driven Web/API). Read the SQLAlchemy 2.0 practical guide alongside for the design of SQLAlchemy 2.0's Mapped / select() / Session themselves.

2. Setup: The Factory and db = SQLAlchemy(model_class=Base)

Onto the bare extension in extensions.py → bind with init_app pattern established in the large-app structure guide, load Flask-SQLAlchemy correctly.

2.1 The Current Canonical Form of Passing DeclarativeBase

The creation method Flask-SQLAlchemy 3.1's official docs recommend is the form of passing a Base inheriting SQLAlchemy 2.0's DeclarativeBase to model_class. With this, db.Model behaves as the 2.0 declarative base, and the typed models described later write naturally.

# 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 を知らない

Defining Base yourself lets you consolidate common mixins (auto-adding created_at / updated_at, etc.) and type-annotation defaults here. This is not "a future extension" but a common foundation you'll surely want soon in production, so it's worth carving out Base from the start.

2.2 Configure in the Factory, Bind with 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 takes a string or a SQLAlchemy URL object. In production, you naturally flow this value in from an environment variable (FLASK_SQLALCHEMY_DATABASE_URI='postgresql+psycopg://...', see Flask production-operations guide §4's from_prefixed_env). Note that migrate.init_app(app, db) passes db as its 2nd argument, unlike extensions other than Flask-SQLAlchemy.

2.3 create_all Only "Inside an Application Context"

db.create_all(), which physically creates tables, requires an active application context. Because an app made in the factory has no automatic context outside a request, you need to push it by hand.

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

Forget this and it falls over with RuntimeError: Working outside of application context (for the context mechanism, see the context deep-dive and §5).

⚠️ Don't use create_all for production schema management: db.create_all() is "just creates tables that don't yet exist." It does none of adding columns, changing types, adding constraints, or dropping on existing tables. That is, it has no means to evolve a once-created table. create_all is permitted only for a throwaway test DB or an initial local prototype. The production schema management is, uniquely, §7's Flask-Migrate (Alembic), and writing create_all in a production-deploy script is a one-way ticket to the trap of never being able to change the schema again.

3. Model Definition: SQLAlchemy 2.0's Typed Style

With Flask-SQLAlchemy 3.1 + SQLAlchemy 2.0, write models in the typed style. A form using Mapped[...] type annotations and mapped_column(), the current official documentation teaches only this style.

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

Let me read precisely what's happening here.

• Mapped[int] / Mapped[str]: declare the column's Python type with a type annotation. The type checker (mypy / Pyright) comes to understand user.email as str, erasing the db.Column("email", String)-era problem of "the type isn't conveyed at all."
• mapped_column(...): specify the column's DB-side constraints (primary_key / unique / nullable, etc.). Mapped[str] (not Optional) maps to NOT NULL, and Mapped[str | None] to NULL-allowed — that the type drives the DB constraint is the crux of 2.0.
• Omitting mapped_column() as in email: Mapped[str] infers the column from the type annotation alone (this suffices for a simple column needing no extra constraints).

💡 db.Column is not "deprecated" but "an alias for the old style": it's easily misunderstood, but db.Column isn't removed, nor does it emit a deprecation warning. In SQLAlchemy 2.0, db.Column still exists as a legacy alias. However, the current official documentation teaches only the typed Mapped / mapped_column style, and writing new code this way is the proper path. You needn't immediately rewrite all your existing db.Column models, but align newly-written models to the typed style — this is the policy that doesn't add debt.

Full-scale modeling like relationships, composite constraints, indexes, and hybrid properties is SQLAlchemy 2.0's domain, not Flask-specific, so I leave the deep-dive to the SQLAlchemy 2.0 practical guide. This article shows only a minimal example of stretching a relationship bidirectionally with 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")

That tenant_id has index=True and (tenant_id, created_at) has a composite index is the foreshadowing of §8's multi-tenant list query and the N+1 countermeasure.

4. Queries: The 2.0 Style (session.execute(select(...))) Is the Right Way

This is the biggest change from SQLAlchemy 1.x. Model.query / session.query() is officially stated as "legacy in SQLAlchemy." The current right way is db.session.execute(db.select(...)).

4.1 Basics: select → execute → scalars

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

Assemble the query with db.select(User), run it with db.session.execute(...), and take out the column of "entities (User)" rather than "rows (Row)" with .scalars(). Without .scalars(), tuple rows like (User,) are returned, so when you want ORM entities, always go through .scalars().

4.2 Flask-SQLAlchemy's *_or_404 Helpers

In the Web context, "404 if not found" is frequent. Flask-SQLAlchemy provides integration helpers that make this one line.

# 主キーで取得、無ければ 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))

These call abort(404) internally, so they ride directly on the JSON error handler set up in §6 (or the handler of the error-handling / observability guide). You avoid scattering if user is None: abort(404) across views.

4.3 Pagination: db.paginate

For a list API, use db.paginate. Pass a select and a Pagination object is returned, holding the page number, count, and whether there are previous/next pages.

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 are read by default from the request's query string (?page=2&per_page=20). This erases the boilerplate of "parse the query string by hand and compute the offset." Wiring into an actual list endpoint is shown in §8.

⚠️ Don't use Model.query in new code just because it still works: Flask-SQLAlchemy still provides User.query for backward compatibility, but its substance is the legacy Query interface on the SQLAlchemy side. Writing like User.query.filter_by(...).first() does work, but compared to the 2.0 select-based way, its type inference is weaker, it has no migration path to async sessions, and it deviates from SQLAlchemy's official future direction. Unify new code on db.session.execute(db.select(...)), and don't use Model.query except when touching existing code — this is the discipline of the 2.0 era.

5. The Application Context and Per-Request Sessions

Without settling Flask-SQLAlchemy's behavior as "magic," precisely understand the relationship between context and session. This is the dividing point that prevents accidents like DetachedInstanceError and "committed but not reflected" in production.

5.1 The Official's "Session Scoped to the Context"

Let me quote the official's 2 most important facts directly.

1. "An active Flask application context is required to issue queries and to access db.engine / db.session."
2. "Because sessions are scoped to the context, they are cleaned up appropriately after each request or CLI command."

That is, db.session is a scoped session tied to the application context currently being processed (≒ 1 request, or 1 CLI command). An independent session is allocated per request, and when the request ends, Flask-SQLAlchemy auto-cleans up that session (remove). You don't need to write a teardown — this means, for the DB, the extension takes off your hands the "write get_db + teardown_appcontext yourself" pattern shown in Flask production-operations guide §5.

5.2 The Responsibility for commit / rollback Is Yours

What's automatic is the session's lifetime management (allocation and cleanup), not the transaction's commit. Always commit writes explicitly, and rollback on failure.

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

Stage changes with db.session.add(obj) / db.session.delete(obj), and confirm with db.session.commit(). Neglect rollback on an exception and the session enters a "broken (a failed transaction remains)" state, and subsequent queries chain-fail with PendingRollbackError.

💡 At the end of a request, Flask-SQLAlchemy doesn't auto-commit: a point easy to misunderstand. What the extension does is session.remove() (session cleanup), not commit. Changes not committed are discarded at the end of the request. Almost all "I add-ed but it isn't in the DB" accidents are a forgotten commit. In a write-type view, place the commit on a path that's always traversed.

5.3 When to "Push the Context by Hand"

During request processing or flask CLI command execution, Flask auto-pushes the context, so you needn't be conscious of it. Write with app.app_context(): by hand only when touching the DB outside a request/CLI.

The official's guidance on testing is clear — "only push a context exactly where and for how long it's needed for each test." Rather than wrapping the whole test in a giant app_context(), wrap only the block needing DB access, pinpoint. How to assemble tests I leave to Flask production-operations guide §9 and each test article. For the push/pop mechanism of the context itself, see the context deep-dive.

6. The Production Connection Pool and Configuration

It works with SQLite in development, but in production (PostgreSQL on RDS/Aurora, often via PgBouncer), the connection-pool configuration governs stability. Flask-SQLAlchemy controls this via SQLALCHEMY_ENGINE_OPTIONS.

6.1 SQLALCHEMY_ENGINE_OPTIONS: A Direct Line to create_engine

SQLALCHEMY_ENGINE_OPTIONS is a dict passed as-is to SQLAlchemy's create_engine(...) that Flask-SQLAlchemy calls internally. The main pool-related keys are these.

The minimal, effective setting I use in production is this (a recommended example, not a verbatim official quote — the appropriate values depend on the workload and the DB-side settings).

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

pool_pre_ping and pool_recycle are decisively important in an environment where a load balancer, PgBouncer, or serverless DB silently severs idle connections. Without setting these, you grab a "connection already dead on the other side" left in the pool and step on OperationalError: server closed the connection unexpectedly. pool_pre_ping=True sends a lightweight ping at checkout to reject this preemptively, and pool_recycle proactively recreates old connections.

💡 Behind PgBouncer / serverless, make the SQLAlchemy-side pool "thin": interpose PgBouncer (transaction mode), RDS Proxy, or serverless Postgres and connection pooling is handled by the proxy side, so holding a thick pool on the SQLAlchemy side creates a double pool and connections overflow. In this case, make pool_size small, or consider NullPool, and keep pool_pre_ping enabled. The proxy-premise pool design (including the handling of prepared statements and the difference between transaction/session modes) is deep-dived in the PostgreSQL connection-pooling guide. The Flask-SQLAlchemy side only decides "what to pass to ENGINE_OPTIONS"; the actual pool behavior is SQLAlchemy/proxy's domain.

6.2 SQLALCHEMY_TRACK_MODIFICATIONS Must Be Off

SQLALCHEMY_TRACK_MODIFICATIONS is disabled by default since Flask-SQLAlchemy 3.0. This is an intentional default, and you must not enable it. In the official words — "this adds non-trivial overhead to every session. It's recommended to use SQLAlchemy's ORM events directly."

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

If you genuinely need object-change notifications, use SQLAlchemy's native ORM events (before_flush, etc.), not this feature.

6.3 SQLALCHEMY_BINDS: Multiple DBs and Read Replicas

When handling multiple databases in one app, use SQLALCHEMY_BINDS. Declare a separate URI per key (bind key), and specify the bind in models or queries.

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",
}

To fix a bind per model, use __bind_key__; to route per query, specify the bind at execution time. Routing to a read replica requires the design judgment of "flow only read-only queries to the replica, and keep writes and reads sensitive to replica lag on the primary," so SQLALCHEMY_BINDS is the entrance to the wiring, and the routing strategy itself needs separate design.

7. Flask-Migrate: Evolve the Production Schema with Alembic

create_all (§2.3) is usable only up to prototyping, and the production schema evolution is, uniquely, Flask-Migrate (Alembic). The official definition is this — "Flask-Migrate is an extension that configures Alembic to work properly with Flask and Flask-SQLAlchemy applications… everything is handled by Alembic." Flask-Migrate itself is a thin wrapper; the substance of migration is Alembic.

7.1 Setup and the 4 CLIs

Once migrate.init_app(app, db) (§2.2) is done, the flask db subcommands are usable. In production operation, you effectively use these 4.

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

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

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

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

The migrations/versions/*.py generated by migrate -m "..." must always be committed to Git. This becomes the single source of truth for schema changes, and the guarantee that the same-ordered migrations are applied across all environments (dev, staging, prod).

7.2 autogenerate Is a "Draft" — Always Review It

This is the point that most causes accidents in production. flask db migrate's auto-generation (autogenerate) is convenient but not perfect. Let me quote the official warning directly — "the migration script needs to be reviewed and edited. Alembic isn't able to detect all changes… Alembic currently cannot detect table-name changes, column-name changes, or anonymously-named constraints."

What happens concretely.

• It misdetects a column "rename" as "drop + add": Alembic generates a old_name → new_name rename as drop_column("old_name") + add_column("new_name"). Run this as-is and that column's data is all lost. A rename needs to be hand-rewritten to op.alter_column(..., new_column_name=...).
• It doesn't detect a table-name change: similarly it becomes "drop + create new," and data is lost.
• It doesn't detect anonymously-named constraints: changes to nameless constraints (an unnamed UniqueConstraint(...), etc.) can't be tracked.

⚠️ Don't commit/apply autogenerate without review (the most frequent cause of production accidents): before git add-ing flask db migrate's output, always eyeball-review the generated upgrade() / downgrade(). Especially the 3 points "is a drop_column showing," "did a rename intent become a drop+add," and "does downgrade() correctly reverse upgrade()" I confirm every time. In my operation, I made reviewing autogenerated migrations a mandatory item of the same PR as normal code review. Because a schema change, once it flows to production, is costly to roll back, the review discipline here divides the life and death of the production DB.

7.3 Run Migrations in CI/CD and Containers

In production, on every deploy, auto-run flask db upgrade to advance the schema. In a container environment like ECS (Fargate), the standard is to run flask db upgrade as a dedicated one-shot task "before" app startup.

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

⚠️ Don't have multiple workers/tasks run upgrade simultaneously: if you design each Gunicorn worker or scaled-out multiple Fargate tasks to each call flask db upgrade at startup, the same migration conflicts and gets stuck on double-application or a lock. Run a migration only once as one deploy step (a single pre-task/job), and have the app's worker群 merely start against the migrated DB.

To apply a schema change without downtime (coexist the existing app and the new schema), you need the expand-and-contract pattern that meshes the order of applying upgrade with the deploy order of the code. The multi-stage-deploy design of "add a column → create a state where both new and old code can read it → drop the old column later" exceeds Flask's domain, so I leave the deep-dive to the Alembic zero-downtime migration guide. This article's flask db upgrade handles "how to run it," and that article handles "in what order to apply it safely."

8. A Production Worked Example: A Multi-Tenant List + Create Endpoint

Let me integrate the elements so far into a realistic endpoint of a multi-tenant B2B SaaS. I land tenant isolation, pagination, filtering, the N+1 countermeasure, and commit/rollback into one 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

Let me organize the production discipline put into this implementation.

• Guarantee tenant isolation structurally with "scope every select by tenant_id." Always apply where(Order.tenant_id == tenant_id) at the entrance. Forget this and you get the heaviest accident in a SaaS — another tenant's data becomes visible. Rather than relying only on the app layer, doubly binding at the DB layer with PostgreSQL's RLS is robust, but scoping all queries at the app layer first is the minimum line.
• Crush N+1 with selectinload. Lazy-load each order.items in a list and N+1 queries fire for N rows. With options(selectinload(Order.items)), load items in bulk with one additional query, so len(o.items) doesn't fire an additional query inside the loop (for the details of N+1 and eager-loading strategies, the SQLAlchemy 2.0 practical guide).
• Validate external input with an allowlist (status's allowed values, items's type). request.args / request.get_json() are untrusted input. Full-scale schema validation I lean to marshmallow — the layered separation of serializing/deserializing models at the API boundary is detailed in Designing a production REST API with marshmallow × Flask × SQLAlchemy. This article handles "how to hit the DB," and that article handles "how to validate/shape at the boundary."
• db.paginate(stmt, max_per_page=100) binds the per_page upper bound, preventing a DoS-ish request like ?per_page=100000.
• Writes are add → commit, exceptions are rollback (§5.2). Place the commit on an always-traversed path, and on failure rollback the session before returning 500.

9. Production DB Design Checklist

Let me summarize the key points so far into a practical checklist to pass before review.

Summary: Design the Data Layer with "Automatic Lifetime" and "Explicit Confirmation"

Because Flask itself has no data layer, the design of binding Flask-SQLAlchemy and Flask-Migrate correctly decides production quality. Let me re-list this article's discipline.

1. Create with db = SQLAlchemy(model_class=Base), the bare extension in extensions.py → init_app in the factory. create_all only inside with app.app_context():, and limited to prototypes/tests.
2. Write models in SQLAlchemy 2.0's typed style (Mapped / mapped_column / relationship). db.Column remains as a legacy alias, but align new ones to the typed style.
3. Queries are db.session.execute(db.select(...)) — the right way. Model.query is legacy. Shorten the Web boilerplate with get_or_404 / one_or_404 / db.paginate.
4. db.session's lifetime is scoped to the context and auto-cleaned-up, but the responsibility for commit / rollback is yours. No auto-commit at request end.
5. The production pool rejects dead connections with pool_pre_ping / pool_recycle (mandatory behind PgBouncer/serverless). TRACK_MODIFICATIONS off, read replicas with SQLALCHEMY_BINDS.
6. Schema evolution with Flask-Migrate (Alembic). autogenerate is a draft, and renames/constraints aren't detected, so always review. Run flask db upgrade once at deploy.
7. Multi-tenant scopes every query by tenant_id, crushes N+1 with eager loading, and hardens boundaries with allowlist + marshmallow.

These are designs that work from the first table, not "once it gets large." Survey the whole picture back at the Flask production-operations guide, the ORM modeling at the SQLAlchemy 2.0 practical guide, and zero-downtime migration at the Alembic zero-downtime migration guide — Flask's data layer is, with these 3, complete in "how to load it, how to model it, how to evolve it."