失敗しないDBマイグレーション：FastAPI×SQLAlchemy×Alembicで進化し続けるスキーマ運用――自動生成、手動調整、データ移行、トランザクション、テストとCI

要約（最初に全体像）

Alembicでスキーマをバージョン管理し、FastAPIとSQLAlchemyの変更を安全に反映する手順をまとめます。
自動生成（autogenerate）と手動編集の合わせ技、複数環境の切替、データ移行スクリプト、トランザクションの扱いを整理します。
ローカルから本番までのチェックリスト、テストとCIへの組み込み方、運用で起きがちな落とし穴も網羅します。

誰が読んで得をするか

学習者Aさん（個人開発）
モデルを直したら手元では動くが、他の環境で崩れる。マイグレーションを正しく回したい。
小規模チームBさん（受託3名）
仕様変更が多く、DB差分を手作業で追っている。自動生成とレビューの流れを標準化したい。
SaaS開発Cさん（スタートアップ）
ステージングと本番で安全にロールアウトし、トラブル時には即ロールバックしたい。

アクセシビリティ評価

章ごとに短い段落・箇条書きで構造化し、コードは固定幅でコメントを控えめに記載。
初出語には簡潔な説明を添え、手順は番号付きで順序を把握しやすくしました。
目標レベルはAA相当です。

1. 前提とゴール

ORMはSQLAlchemy 2.xスタイル、スキーマ移行はAlembicを使用。
環境はSQLiteでも学習できますが、本番相当はPostgreSQLなどのRDBを想定。
ゴールは、モデル変更→差分生成→レビュー→適用→検証までを定型化し、誰がやっても同じ結果になる状態です。

判断ポイント

本番は必ずトランザクション対応のDBを使う。
自動生成は便利ですが、インデックス名・デフォルト値・サーバ側関数などは手動調整が要ります。

2. 初期化：Alembicを導入して動かす

2.1 依存パッケージ

SQLAlchemy>=2.0
alembic
psycopg[binary]  # PostgreSQLを使う場合の例

2.2 ディレクトリ構成（例）

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

2.3 初期化コマンド

alembic init alembic

生成物のポイント

alembic.ini にDB URLのひな形が入る。
alembic/env.py がマイグレーション実行時のエントリ。ここでSQLAlchemyのBaseを読み込み、autogenerateの対象テーブルを認識させます。

3. アプリ側のDB設定とBase

3.1 設定

# app/settings.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
    database_url: str = "postgresql+psycopg://user:pass@localhost:5432/appdb"
    class Config:
        env_file = ".env"
def get_settings() -> Settings:
    return Settings()

3.2 セッションとBase

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

class Base(DeclarativeBase): pass

def get_engine(url: str):
    return create_engine(url, pool_pre_ping=True)

def make_session_factory(url: str):
    return sessionmaker(bind=get_engine(url), autoflush=False, autocommit=False)

3.3 モデル例

# app/models.py
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Integer, DateTime, func
from app.db import Base

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True, nullable=False)
    body: Mapped[str] = mapped_column(String, nullable=False)
    created_at: Mapped["DateTime"] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)

4. env.py をFastAPIプロジェクトとつなぐ

alembic/env.py に、アプリのBaseとDB URLを読み込む処理を書きます。

# alembic/env.py（抜粋）
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# 1) アプリのBaseと設定を import
import os, sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from app.db import Base
from app.settings import get_settings

config = context.config
if config.get_main_option("sqlalchemy.url") in (None, "", "sqlite:///"):
    config.set_main_option("sqlalchemy.url", get_settings().database_url)

target_metadata = Base.metadata

fileConfig(config.config_file_name)

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        compare_type=True,      # 型変更の検出
        compare_server_default=True  # サーバデフォルトの検出
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section), prefix="sqlalchemy.", poolclass=pool.NullPool
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True
        )
        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 を有効化すると差分検出が安定します。
URLは環境変数や.envから取得し、alembic.iniのデフォルトを上書きします。

5. 初回マイグレーションを作る

5.1 ベース作成

alembic revision --autogenerate -m "init"
# 生成後に確認→適用
alembic upgrade head

alembic/versions/<rev>_init.py が生成されます。テーブルやインデックスが意図通りかをレビューしてから適用します。

5.2 よくある手動調整

文字列長の変更はDBによって注意（縮小はデータ喪失の可能性）。
server_default=func.now() のような式はDB方言により差分が出やすい。必要に応じて手書き指定。
複合インデックスやユニーク制約は命名規約を決めて明示します。

6. 変更を加えて差分を回す

例として、Articleにslug列を追加しユニーク制約を持たせます。

6.1 モデル変更

# app/models.py（追記）
from sqlalchemy import UniqueConstraint
class Article(Base):
    __tablename__ = "articles"
    # 既存の列は省略
    slug: Mapped[str] = mapped_column(String(200), nullable=False, unique=True)
    __table_args__ = (UniqueConstraint("slug", name="uq_articles_slug"),)

6.2 差分作成→レビュー→適用

alembic revision --autogenerate -m "add slug to articles"
alembic upgrade head

生成されたスクリプトに、op.add_column や op.create_unique_constraint が並びます。名前・順序・ダウン操作の妥当性を確認します。

7. データ移行（データマイグレーション）の書き方

スキーマ追加に伴い既存行へ初期値や変換が必要な場合は、同じリビジョンでデータ操作を入れます。

# alembic/versions/xxxx_add_slug.py（抜粋）
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column("articles", sa.Column("slug", sa.String(length=200), nullable=True))
    # 一時的にNULL許可→既存データに埋める→非NULLに変更する手順が安全
    conn = op.get_bind()
    articles = conn.execute(sa.text("SELECT id, title FROM articles")).mappings().all()
    for row in articles:
        slug = row["title"].lower().replace(" ", "-")
        conn.execute(sa.text("UPDATE articles SET slug=:s WHERE id=:i"), {"s": slug, "i": row["id"]})
    op.alter_column("articles", "slug", nullable=False)
    op.create_unique_constraint("uq_articles_slug", "articles", ["slug"])

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

ポイント

既存データがあるときは、列追加→データ埋め→制約付与の順で段階適用。
大量更新は長時間ロックの原因。バッチで分割したり、業務時間外に行う計画を立てます。

8. トランザクションとロールバック

PostgreSQLなどでは、各マイグレーションは基本的にトランザクション内で実行され、失敗時は自動ロールバックされます。
テーブル再作成やロングロックを伴う操作（巨大な列型変更など）は、事前に影響調査をします。
SQLiteはDDLのトランザクション特性が弱く、autogenerateの結果もDB制約により変わることがあります。本番相当のRDBで検証するのが安全です。

9. 複数環境の運用（dev/stg/prod）

dev：自由にマイグレーション作成・適用。ダミーデータで検証。
stg：本番同等のDBで適用テスト。所要時間とロック時間を計測。
prod：メンテナンス時間帯やローリング配備と連携し、監視とアラートを見ながら実行。

Tips

alembic current で現在のバージョン、alembic history で履歴を確認。
緊急時の戻しは alembic downgrade -1 などを使いますが、データ損失を伴うダウングレードは原則避けます。前方修正の方針を基本にします。

10. ブランチとマージの扱い

複数の開発ラインで同時にリビジョンを切ると、バージョングラフが分岐します。

競合を避けるには、主ブランチへ頻繁に取り込み、alembic revision --autogenerate -m "merge heads" で統合する。
統合リビジョンでは、どちらの差分も順序正しく適用されるように確認します。
命名規約とレビューを厳密にし、開発者間で生成タイミングを合わせるとトラブルが減ります。

11. FastAPIアプリの起動時に自動適用してよいか

ローカル開発では「起動時に upgrade head」は便利です。
本番ではヒューマンレビューと手動実行、あるいはCI/CDパイプラインの明示ステップで適用するのが安全です。
コンテナ起動フック（Entrypoint）で alembic upgrade head を実行し、失敗なら起動を止める構成は、IaCと相性がよいです。

サンプル（シェル）：

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

12. テストに組み込む

単体テストでは、テストDBを作成→alembic upgrade head→テスト→ロールバックで環境を毎回クリーンに。
既存データを伴うマイグレーションは、スナップショット用のダミーデータを用意し、upgrade が成功するかと結果の整合を検証します。
失敗時に再実行しても結果が壊れないこと（再入可能性）を確認します。

pytest例（概略）：

import subprocess, os
def setup_module():
    os.environ["DATABASE_URL"] = "postgresql+psycopg://user:pass@localhost:5432/testdb"
    subprocess.check_call(["alembic", "upgrade", "head"])

13. CI/CD の流れ

PRで alembic revision --autogenerate を含むコミットがあれば、CIで alembic upgrade head を検証。
ステージングにデプロイしたあと、マイグレーションの実行時間とログを保存し、本番前にアラート条件を緩和しておく。
リリース手順に「DBマイグレーション→アプリのロールアウト」の順序を明示します。

14. 運用の小さなコツ

命名規約：add_<col>_to_<table>、create_<table>、drop_<col>_from_<table> のように目的が明快だと履歴が読みやすい。
大きな変更は複段階に分割し、一度にやり過ぎない。
重いインデックス作成は並行作成オプション（PostgreSQLの CONCURRENTLY）を検討。ただしAlembicでは手動SQLが必要になります。
本番データ量に近いステージングで必ず計測する。

15. よくある落とし穴と回避策

症状	原因	対策
autogenerateが検出しない	Baseのimport漏れ、`target_metadata`未設定	env.pyでBase.metadataを確実に渡す
文字列長縮小で失敗	既存データが長い	検出クエリで超過行を洗い出し、トリム方針を合意してから段階移行
サーバデフォルトの差異で差分が出続ける	方言差、関数表現の違い	`compare_server_default`を活かしつつ、手書きで関数名を固定
ロックが長引く	巨大テーブルの再書き込み	オンライン手法の検討、分割適用、夜間適用、インデックス並行作成
ダウングレードで壊れる	破壊的操作	前方修正を基本。どうしても戻す必要がある場合は事前にバックアップ

16. 具体例：タグテーブルの導入と関連付け

記事とタグの多対多を例に、段階的にマイグレーションを作成します。

tags テーブル作成
中間 article_tags 作成（複合主キー）
既存記事にタグを付与（データ移行）
必要なインデックス追加

モデル変更（抜粋）：

# app/models.py（追記例）
from sqlalchemy import Table, Column, ForeignKey
from sqlalchemy.orm import relationship

article_tags = Table(
    "article_tags",
    Base.metadata,
    Column("article_id", ForeignKey("articles.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)

class Article(Base):
    __tablename__ = "articles"
    # 既存の列は省略
    tags = relationship("Tag", secondary=article_tags, backref="articles")

差分生成→手動で中間テーブルにインデックスやondeleteの方針を確認し、upgrade に必要ならデータ投入SQLを加えます。

17. ロードマップ

Alembicの初期化とenv.pyの整備。
最初のテーブル群をautogenerateで作成し、レビューして適用。
モデル変更→差分生成→手動調整→適用の反復をチームの標準に。
データ移行を安全な順序で書き、テストとステージングで時間・ロックを計測。
CI/CDに組み込み、ブランチ統合や緊急対応の手順を明文化。

参考リンク

Alembic
SQLAlchemy
- SQLAlchemy 2.0 ORM
- Migration from 1.x to 2.x
PostgreSQL
- PostgreSQL Documentation
FastAPI
- FastAPI SQL Databases

まとめ

Alembicでスキーマをバージョン管理し、差分は自動生成を起点に手動で仕上げる。
既存データがある変更は段階適用とデータ移行をセットで設計し、トランザクションとロックの影響を必ず計測する。
テストとステージング、本番の三段で同じ手順を踏めるようにし、CI/CDにチェックを組み込む。
ブランチ統合や命名規約を明確にし、履歴が読める形を保つ。これで、モデルの進化がそのまま安全なリリースに直結します。