FastAPI×SQLAlchemy×AlembicのDBマイグレーション実務ガイド：安全な変更、データ移行、CI

FastAPIのアプリでSQLAlchemyのモデルを変更しても、データベースのテーブル構造は自動では変わりません。

Alembicで変更履歴を作り、内容をレビューしてから各環境へ適用すると、アプリのコードとデータベースの食い違いを減らせます。

ただし、autogenerate（差分生成）は完成した移行手順ではなく、モデルと現在のDBを比較して作る候補です。

既存データへの値の補完、列名の変更、ロック時間、復旧方法は、開発者が設計して確認する必要があります。

DBマイグレーションで管理するもの

DBスキーマは、テーブル、列、型、インデックス、外部キーなど、データを保存する構造の定義です。

マイグレーションは、スキーマをある状態から次の状態へ変更する手順です。

リビジョンは、その手順を履歴として識別する単位であり、AlembicではPythonファイルとして保存されます。

安全な運用では、モデル変更から本番反映までを次の流れにそろえます。

Alembicを使う標準的な変更手順

段階	作業	確認すること
設計	モデルと移行方針を決める	既存データ、互換性、ロック、復旧方法
作成	リビジョン候補を生成する	意図しない削除や制約漏れがないか
検証	使い捨てのDBとステージングへ適用する	結果、所要時間、アプリとの互換性
反映	承認済みの同じリビジョンを本番へ適用する	ログ、監視、停止条件

Alembicの操作を初めて試す場合は、先にFastAPI、SQLAlchemy、AlembicのDBマイグレーション入門で作成、適用、ロールバックの基本を確認できます。

前提となる構成

この記事ではSQLAlchemy 2.xの宣言型モデルとAlembicを使い、開発用にはSQLiteも選べるものの、本番相当の検証には実際の運用先と同じ種類のRDBを使う構成を想定します。

FastAPIはAPI層を担当し、AlembicがSQLAlchemyのメタデータとDBスキーマを比較して変更履歴を管理します。

SQLAlchemy>=2.0
alembic
pydantic-settings
psycopg[binary]  # PostgreSQLを使う場合

ディレクトリは次のように分けると、アプリの設定、モデル、マイグレーション履歴の役割が見えやすくなります。

fastapi-app/
├─ app/
│  ├─ main.py
│  ├─ db.py
│  ├─ models.py
│  └─ settings.py
├─ alembic/
│  ├─ env.py
│  ├─ script.py.mako
│  └─ versions/
└─ alembic.ini

alembic init alembic

このコマンドは設定ファイルと履歴用ディレクトリを作りますが、アプリのモデルや接続先との連携はまだ完了していません。

SQLAlchemyのモデルと接続設定

DBの接続先はソースコードへ固定せず、環境変数またはローカルの設定ファイルから読み込みます。

Pydantic Settingsで接続先を読む

# app/settings.py
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    database_url: str

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )


@lru_cache
def get_settings() -> Settings:
    return Settings()

.envにはDATABASE_URLを設定し、認証情報をリポジトリへコミットしない運用にします。

エンジンとBaseを定義する

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.settings import get_settings


class Base(DeclarativeBase):
    pass


engine = create_engine(
    get_settings().database_url,
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
)

型注釈とDB型を対応させる

# app/models.py
from datetime import datetime

from sqlalchemy import DateTime, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base


class Article(Base):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    body: Mapped[str] = mapped_column(Text, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

Mapped[datetime]にはPythonの日時型を指定し、mapped_column()でDB側の型とデフォルト値を定義します。

SQLAlchemyによるAPI実装も確認したい場合は、FastAPI、SQLite、SQLAlchemyで作るCRUD APIの例が参考になります。

Alembicのenv.pyをアプリへ接続する

alembic/env.pyには、接続先とBase.metadataを渡します。

モデルを別モジュールへ分けている場合は、それらを読み込んでからメタデータを参照しないと、Alembicがテーブルを認識できません。

# alembic/env.py（要点）
from logging.config import fileConfig

from alembic import context
from sqlalchemy import create_engine, pool

from app import models  # noqa: F401
from app.db import Base
from app.settings import get_settings


config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def get_url() -> str:
    return get_settings().database_url


def run_migrations_offline() -> None:
    context.configure(
        url=get_url(),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
        compare_server_default=False,
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = create_engine(get_url(), poolclass=pool.NullPool)

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=False,
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

compare_type=Trueは型の差分を比較しますが、方言固有の細部まで常に判断できるわけではありません。

compare_server_default=Trueによるデフォルト値の比較も単純な場合には役立ちますが、表現差による不要な差分が出ることがあるため、対象DBで結果を確かめてから有効にします。

最初のリビジョンを作成してレビューする

alembic revision --autogenerate -m "init"
alembic upgrade head

二つのコマンドを続けて機械的に実行するのではなく、生成されたリビジョンを確認してからupgradeへ進みます。

差分生成のレビューでは、少なくとも次を確認します。

• 意図しないdrop_tableやdrop_columnが含まれていないか。
• 列の型、NULL可否、サーバーデフォルトが設計どおりか。
• インデックス、外部キー、一意制約に一貫した名前が付いているか。
• 列名やテーブル名の変更が「削除と追加」として出ていないか。
• downgrade()が運用方針に合い、データを失う操作が明示されているか。
• スキーマ変更の前後に必要なデータ更新が抜けていないか。

Alembicのautogenerateは列名やテーブル名の変更を意図どおりに判定できないため、生成結果をそのまま本番へ適用する前提にはできません。

既存データを守る段階移行

既存のarticlesテーブルへ、NULLを許可せず一意でなければならないslug列を追加する場面を考えます。

最終形だけをモデルへ書いて差分を生成すると、既存行に値がないまま非NULL制約を付けようとして失敗する可能性があります。

モデルには最終形を書く

from sqlalchemy import String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column


class Article(Base):
    __tablename__ = "articles"
    __table_args__ = (
        UniqueConstraint("slug", name="uq_articles_slug"),
    )

    # 既存の列は省略
    slug: Mapped[str] = mapped_column(String(200), nullable=False)

unique=Trueと明示的なUniqueConstraintを重ねず、一つの名前付き制約にそろえます。

移行は追加、補完、検証、制約の順にする

# alembic/versions/xxxx_add_article_slug.py
from alembic import op
import sqlalchemy as sa


def upgrade() -> None:
    op.add_column(
        "articles",
        sa.Column("slug", sa.String(length=200), nullable=True),
    )

    connection = op.get_bind()
    connection.execute(
        sa.text(
            """
            UPDATE articles
            SET slug = 'article-' || CAST(id AS VARCHAR)
            WHERE slug IS NULL
            """
        )
    )

    missing_count = connection.execute(
        sa.text("SELECT COUNT(*) FROM articles WHERE slug IS NULL")
    ).scalar_one()
    if missing_count:
        raise RuntimeError("slug backfill is incomplete")

    op.alter_column(
        "articles",
        "slug",
        existing_type=sa.String(length=200),
        nullable=False,
    )
    op.create_unique_constraint(
        "uq_articles_slug",
        "articles",
        ["slug"],
    )


def downgrade() -> None:
    op.drop_constraint(
        "uq_articles_slug",
        "articles",
        type_="unique",
    )
    op.drop_column("articles", "slug")

このリビジョン例はPostgreSQLを対象としています。

補完SQLの文字列連結はSQLiteでも使えますが、SQLiteの制約変更は後述のbatch modeに合わせて書き換えます。

article-{id}は手順を説明するための暫定値なので、公開URLに使うslugの仕様は重複、文字種、変更可否を含めて別に決めます。

大量の行を一度に更新するとロックやログ量が増えるため、列追加、アプリによる新規行の書き込み対応、バッチ補完、制約付与を複数回のリリースへ分ける方法もあります。

このように旧版と新版が一時的に共存できる形で広げ、データを移し、最後に古い状態を閉じる手順をexpand and contractと呼びます。

トランザクションとロックを別々に考える

Alembicのcontext.begin_transaction()が示すのは論理的なトランザクション境界であり、実際にDDLをロールバックできるかはDB、ドライバー、実行するDDLに依存します。

PostgreSQLの多くのDDLはトランザクション内で扱えますが、トランザクション外で実行すべき操作もあります。

そのため、「失敗すればすべて自動で元に戻る」と仮定せず、ステージングで失敗条件まで試します。

トランザクションで戻せる操作でも、実行中のロックが短いとは限りません。

大きなテーブルの型変更、全行更新、インデックス作成では、次を事前に測ります。

• 対象行数とテーブル容量。
• SQLの実行時間と取得するロック。
• 待機中のクエリとタイムアウトへの影響。
• 中断した場合の再開方法。
• バックアップから復旧する場合の所要時間。

PostgreSQLでインデックスの並行作成を使う場合はトランザクション境界が通常の移行と異なるため、Alembicのautocommit blockと方言固有オプションを含めて個別に検証します。

一つのリビジョンの途中で自動コミットが必要になる操作は、それ以前の処理が先に確定する可能性も踏まえて小さく分けます。

ダウングレードを復旧計画と混同しない

alembic downgrade -1は直前のリビジョンを戻す手段ですが、削除した列のデータや変換前の値まで必ず復元できるわけではありません。

破壊的な変更では、直前の状態へ戻すより、問題を修正する新しいリビジョンを前へ適用するほうが安全な場合があります。

復旧計画には、ダウングレードだけでなく、アプリの旧版との互換性、バックアップ、リストア手順、切り戻しの判断者を含めます。

開発、ステージング、本番で同じ履歴を使う

環境ごとにリビジョンを作り直すと、どの変更が適用されたか追えなくなります。

開発環境で作成してレビューした同じリビジョンファイルを、ステージング、本番の順に適用し、環境ごとの差はDB URLや認証情報へ限定します。

• 開発：使い捨てのデータでも移行を繰り返し、生成内容をレビューします。
• ステージング：本番に近いDBとデータ量で、所要時間、ロック、アプリの互換性を確認します。
• 本番：承認済みのリビジョンだけを適用し、ログと監視を見ながら停止条件に従います。

alembic current
alembic heads
alembic history

currentはDBへ適用済みのリビジョン、headsはコード上の先端、historyは履歴を確認するために使います。

ブランチが分岐したときの統合

複数の開発ブランチで同じ親リビジョンから別々のリビジョンを作ると、headが複数になります。

これはファイル名の競合ではなく、履歴グラフの分岐です。

alembic heads
alembic merge -m "merge heads" heads

alembic revision --autogenerateはモデル差分から新しい変更候補を作るコマンドであり、分岐したheadを統合するコマンドではありません。

mergeリビジョンは複数の親を一つのheadへまとめますが、両方の変更が同じテーブルや制約へ干渉する場合は、適用順と整合性を別途レビューします。

アプリ起動時の自動適用を本番で避ける理由

ローカル開発では、起動前にalembic upgrade headを実行すると環境をそろえやすくなります。

本番で複数のアプリレプリカが同時に起動する構成では、それぞれが同じマイグレーションを実行すると競合する可能性があります。

本番では、アプリの起動処理から分けた一度だけのデプロイジョブまたは承認付きのCI/CDステップとして実行するほうが、実行主体とログを追跡しやすくなります。

set -e
alembic upgrade head
exec uvicorn app.main:app --host 0.0.0.0 --port 8000

この形を使うなら、全レプリカの共通Entrypointではなく、一度だけ動く専用ジョブとして配置します。

マイグレーションをテストする

通常のAPIテストでトランザクションを戻すだけでは、別接続で実行したDDLやデータ補完の正しさまで確認できません。

マイグレーション専用の使い捨てDBを用意し、少なくとも次の経路を試します。

1. 空のDBをheadまで適用できるか確認する。
2. 一つ前のリビジョンへ、境界値を含む既存データを投入する。
3. 対象リビジョンへ更新し、列、制約、データの変換結果を検証する。
4. 運用方針でダウングレードを支える場合だけ、使い捨てDBで戻しと再適用を試す。
5. 同じコードで再度upgrade headを実行し、追加の変更が走らないことを確認する。

FastAPI側の依存性差し替えやテストDBの扱いは、FastAPIの自動テスト入門で詳しく解説しています。

CIで止めたい三つの不整合

CIでは、リビジョンが適用できるか、モデル変更に対応するリビジョンがあるか、更新後のアプリが動くかを別々に確認します。

alembic upgrade head
alembic check
pytest

• alembic upgrade headで、リビジョン列を最初から適用できるか確認します。
• alembic checkで、モデルに未反映の差分候補が残っていないか確認します。
• pytestで、変更後のスキーマに対してアプリの読み書きが成立するか確認します。

alembic checkもautogenerateと同じ比較機構を使うため、列名変更などの検出限界は残ります。

CIの成功だけで本番のロック時間は分からないので、ステージングでの計測も省けません。

SQLiteで検証するときの注意

SQLiteは学習や小規模なテストに便利ですが、PostgreSQLやMySQLと同じALTER TABLE機能やロック特性を持つわけではありません。

Alembicのbatch modeは、必要に応じて新しいテーブルを作り、データを移し、元のテーブルを置き換える方法でSQLiteの変更を扱います。

本番が別のRDBなら、SQLiteだけで成功した結果を本番検証の代わりにはせず、同じ種類と主要バージョンのDBでも移行を試します。

症状から確認箇所を絞る

Alembic運用で起きやすい問題と確認箇所

症状	主な確認箇所	対処の方向
変更が検出されない	モデルのimport、target_metadata、接続先	全モデルを読み込んでからBase.metadataを渡す
削除と追加が生成される	列名またはテーブル名の変更	rename操作へ手動で直し、データ保持を確認する
デフォルト値の差分が繰り返される	DB方言と式の表現	対象DBで比較結果を確認し、比較設定を調整する
非NULL列の追加に失敗する	既存行の値	NULL許可で追加し、補完と検証後に制約を付ける
適用が長時間止まる	全行更新、型変更、インデックス、ロック待ち	計測し、分割移行やDB固有のオンライン手法を検討する
複数のheadができる	並行ブランチのリビジョン	変更の干渉を確認してmergeリビジョンを作る

本番適用前のチェックリスト

• 生成したリビジョンに意図しない削除がない。
• 既存データのNULL、重複、文字数、外部キー違反を事前に調べた。
• アプリの旧版と新版が移行途中のスキーマを扱える。
• 本番に近いデータ量で実行時間とロックを測った。
• 実行者、実行時刻、停止条件、連絡先を決めた。
• ダウングレード、前方修正、バックアップ復元のどれを使うか決めた。
• 監視項目と適用後の確認クエリを用意した。
• 本番で生成せず、レビュー済みのリビジョンを適用する。

公式資料

• Alembic公式ドキュメント
• Alembicのautogenerateと検出範囲
• Alembicのブランチとmergeリビジョン
• Alembicのbatch migration
• SQLAlchemy 2.0 ORM
• Pydantic Settings
• PostgreSQL公式ドキュメント
• FastAPIのSQLデータベース解説

安全なDB変更をチームの手順にする

Alembicを導入する目的は、変更ファイルを作ることではなく、誰が適用しても同じ履歴をたどれる状態を保つことです。

差分生成を出発点として手動レビューを行い、既存データの補完と制約付与を分け、DBごとのトランザクションとロックを検証します。

そのうえで、使い捨てDB、ステージング、本番へ同じリビジョンを順に適用すれば、モデル変更を予測可能なリリース手順へ変えられます。

この記事に関連する株式会社greedenの取り組み

DBマイグレーションの安全性は、設計、テスト、リリース後の運用を一つの手順として整えることで高まります。株式会社greedenは、要件定義から保守運用まで一気通貫でWebシステム開発を支援しています。