# Flask's Data Layer: Designing and Operating a Production DB with Flask-SQLAlchemy 3.x (2.0 Style) and Flask-Migrate > A practical guide to designing and operating Flask SQLAlchemy Migrate at production quality. Explained with real code faithful to the official documentation: Flask-SQLAlchemy 3.1 and SQLAlchemy 2.0's typed Mapped/mapped_column, the 2.0 query session.execute(select), get_or_404・db.paginate, the application context and per-request sessions, the SQLALCHEMY_ENGINE_OPTIONS pool_pre_ping/pool_recycle pool design, the read-replica of SQLALCHEMY_BINDS, and Flask-Migrate (Alembic)'s autogenerate review discipline and CI/CD upgrade. - Published: 2026-06-26 - Author: 友田 陽大 - Tags: Python, Flask, SQLAlchemy, Flask-Migrate, データベース, PostgreSQL, 本番運用, バックエンド - URL: https://tomodahinata.com/en/blog/flask-sqlalchemy-flask-migrate-database-production-guide - Category: Flask in production - Pillar guide: https://tomodahinata.com/en/blog/flask-production-guide ## Key points - Flask-SQLAlchemy 3.1 is an extension that binds SQLAlchemy 2.0's session lifetime to Flask's request/context. Create it with db = SQLAlchemy(model_class=Base), bind with init_app, and run create_all only inside with app.app_context() - Write models in the 2.0 typed style (Mapped/mapped_column/relationship). The right way to query is db.session.execute(db.select(...)) rather than Model.query. Shorten the boilerplate with get_or_404・one_or_404・db.paginate - db.session is scoped to the application context, and Flask-SQLAlchemy auto-tears-down at the end of a request/CLI. Push the context by hand only 'outside a request' — init, create_all, direct DB tests, etc. - In production, set pool_pre_ping/pool_recycle with SQLALCHEMY_ENGINE_OPTIONS to reject dead connections. Mandatory behind PgBouncer or serverless. TRACK_MODIFICATIONS off, separate read replicas with SQLALCHEMY_BINDS - Evolve the production DB not with create_all but with Flask-Migrate (Alembic). Always review autogenerate (renames and constraint names aren't detected), and build the deploy's flask db upgrade into CI/CD. Explained grounded in operations knowledge from a Minister of Economy, Trade and Industry Award-winning B2B SaaS --- ## **Introduction: Flask Has No "Data Layer"** As repeated in the [Flask production-operations guide](/blog/flask-production-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](/blog/flask-production-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](/blog/sqlalchemy-2-typed-orm-production-guide), and the strategy of zero-downtime migration into the [Alembic zero-downtime migration guide](/blog/alembic-zero-downtime-migrations-sqlalchemy). 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. | Concern | Raw SQLAlchemy (yourself) | Flask-SQLAlchemy (takes off your hands) | |---|---|---| | Engine creation | Call `create_engine(url, **opts)` by hand | Auto-create from `SQLALCHEMY_DATABASE_URI` / `SQLALCHEMY_ENGINE_OPTIONS` config | | Session factory | Assemble `sessionmaker` + `scoped_session` | Provides `db.session` (already context-scoped) | | Post-request cleanup | Call `session.remove()` yourself in `teardown` | Auto-cleanup at context end | | Multiple DBs | Bundle multiple engines/sessions by hand | Declaratively with `SQLALCHEMY_BINDS` | | Flask-integration helpers | None | Provides `get_or_404` / `paginate` / `db.Model`, etc. | 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](/blog/sqlalchemy-2-typed-orm-production-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](/blog/flask-application-factory-blueprints-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. ```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 を知らない ``` 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` ```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` 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](/blog/flask-production-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.** ```python 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](/blog/flask-application-request-context-g-current-app-guide) 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. ```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] ``` 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](/blog/sqlalchemy-2-typed-orm-production-guide). This article shows only a minimal example of stretching a relationship bidirectionally with `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") ``` 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` ```python # 全ユーザーを 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()`. | What you want | The 2.0 style | |---|---| | Multiple (iterator) | `db.session.execute(db.select(User)).scalars()` | | Multiple (list) | `db.session.execute(db.select(User)).scalars().all()` | | One (`None` if absent) | `db.session.execute(db.select(User).filter_by(id=1)).scalar_one_or_none()` | | One (raise if absent) | `db.session.execute(db.select(User).filter_by(id=1)).scalar_one()` | | Get by primary key | `db.session.get(User, id)` (no `select` needed) | ### 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. ```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)) ``` 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](/blog/flask-error-handling-logging-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. ```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` 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](/blog/flask-production-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. ```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() ``` **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.** | Scene | Is the context? | Your job | |---|---|---| | Inside a view function | A request context is auto-active | Nothing (`db.session` is usable as-is) | | Inside `@app.cli.command` | An app context is auto-active | Nothing | | Init scripts, `create_all` | Not active | Push with `with app.app_context():` | | A scheduler/worker's startup processing | Not active | Push with `with app.app_context():` | | Tests hitting the DB directly | Depends on the test | Push only where needed | 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](/blog/flask-production-guide) §9 and each test article. For the push/pop mechanism of the context itself, see the [context deep-dive](/blog/flask-application-request-context-g-current-app-guide). --- ## **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. | Key | Role | Meaning in production | |---|---|---| | `pool_size` | The standing connections the pool holds | Match to workers / concurrency | | `pool_recycle` | Reconnect connections older than this many seconds | **Shorter** than the DB/proxy-side idle timeout | | `pool_pre_ping` | Liveness-check with a light ping before checkout | Prevents the accident of grabbing a dead connection | | `pool_timeout` | Seconds to wait for a connection when the pool is exhausted | An upper bound so requests don't pile up | 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). ```python # 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](/blog/postgresql-connection-pooling-pgbouncer-serverless-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."** ```python # 既定でオフだが、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. ```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", } ``` 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. ```bash # ① マイグレーション環境を初期化（最初の一度だけ）。migrations/ を生成 flask db init # ② モデルの変更を検出して、マイグレーションスクリプトを生成 flask db migrate -m "add tenant_id to orders" # ③ マイグレーションを適用（DB を最新スキーマへ前進） flask db upgrade # ④ 直前のマイグレーションを巻き戻す flask db downgrade ``` | Command | Role | When to use | |---|---|---| | `flask db init` | Generate `migrations/` | The project's first time only | | `flask db migrate -m "..."` | **Auto-generate** a script from the diff | During development when you changed a model | | `flask db upgrade` | Advance the DB | **At deploy (CI/CD)** | | `flask db downgrade` | Retreat the DB | Rollback / local verification | 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.** ```bash # デプロイパイプライン（または 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](/blog/alembic-zero-downtime-migrations-sqlalchemy). 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. ```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 ``` 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](/blog/sqlalchemy-2-typed-orm-production-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](/blog/marshmallow-flask-sqlalchemy-rest-api-production-guide). 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. | # | Item | Criterion | This article | |---|---|---|---| | 1 | Created with `db = SQLAlchemy(model_class=Base)` | Passing a `Base` inheriting `DeclarativeBase` | §2.1 | | 2 | Extensions bare in `extensions.py` → `init_app` in factory | `db.init_app(app)` / `migrate.init_app(app, db)` | §2.2 | | 3 | Not using `create_all` for production schema management | Production is Flask-Migrate only; `create_all` for tests/prototypes | §2.3 / §7 | | 4 | Models are typed (`Mapped` / `mapped_column`) | New models in the 2.0 style; no new `db.Column` | §3 | | 5 | Queries are `db.session.execute(db.select(...))` | Not using `Model.query` in new code | §4 | | 6 | `commit` on writes, `rollback` on exceptions | No auto-commit; no missed `rollback` | §5.2 | | 7 | Wrapped DB access outside the context in `app_context()` | Pushing in init/batch/direct tests | §5.3 | | 8 | Set `pool_pre_ping` / `pool_recycle` | Reject dead connections behind proxy/serverless | §6.1 | | 9 | `SQLALCHEMY_TRACK_MODIFICATIONS=False` | Keep the default off; substitute with ORM events | §6.2 | | 10 | Read replica / multiple DBs are `SQLALCHEMY_BINDS` | Separated by bind key | §6.3 | | 11 | Eyeball-reviewed autogenerate | Confirmed drop/add misdetection, renames, downgrade | §7.2 | | 12 | Run `flask db upgrade` once at deploy | Don't have multiple workers/tasks run it simultaneously | §7.3 | | 13 | Scoped every query by `tenant_id` (multi-tenant) | Tenant crossing is structurally impossible | §8 | | 14 | Crushed N+1 with eager loading | Applying `selectinload` etc. in lists | §8 | | 15 | Validated external input at the API boundary | Hardened the boundary with allowlist + marshmallow | §8 | --- ## **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](/blog/flask-production-guide), the ORM modeling at the [SQLAlchemy 2.0 practical guide](/blog/sqlalchemy-2-typed-orm-production-guide), and zero-downtime migration at the [Alembic zero-downtime migration guide](/blog/alembic-zero-downtime-migrations-sqlalchemy) — Flask's data layer is, with these 3, complete in "how to load it, how to model it, how to evolve it."