FastAPI×SQLAlchemy×AlembicのDBマイグレーション入門｜作成、適用、ロールバックの手順

FastAPIアプリの開発では、機能追加に合わせてテーブル、カラム、制約も変わります。変更を手作業だけで進めると、誰がどの順序で何を変えたのかを追いにくくなります。

マイグレーションは、データベースのスキーマ変更をリビジョンというファイルに記録し、順番に適用できるようにする仕組みです。本記事では、SQLAlchemyでモデルを定義し、Alembicで変更履歴を作成して適用する流れを、SQLiteのTodoアプリを例に説明します。

ここで扱うのは、主に同じアプリ内でスキーマを更新する手順です。SQLiteからPostgreSQLへデータを移す作業そのものとは分けて考えます。

この記事でわかること

• FastAPI、SQLAlchemy、Alembicの役割分担
• Alembicの初期設定と自動差分の作り方
• 生成されたリビジョンを確認して適用する手順
• 既存データを守りながら非NULL制約を追加する考え方
• SQLite、複数ブランチ、CIで詰まりやすい点への対処

3つのツールの役割

設定を始める前に、各ツールが担当する範囲を分けておくと理解しやすくなります。

FastAPIのCRUD実装を先に確認したい場合は、FastAPI、SQLite、SQLAlchemyで作るCRUD APIの手順も参照してください。

準備するファイルとパッケージ

最小構成

fastapi-db/
├─ app/
│  ├─ main.py
│  ├─ db.py
│  ├─ models.py
│  └─ schemas.py
├─ alembic/
├─ alembic.ini
└─ .env

alembic/とalembic.iniは、後でalembic initを実行すると作成されます。

必要なパッケージ

python3 -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn sqlalchemy alembic pydantic
# PostgreSQLを使う場合
pip install "psycopg[binary]"

データベースURL

ローカルのSQLiteにはsqlite:///./app.dbを使います。PostgreSQLの接続形式はpostgresql+psycopg://USER:PASSWORD@HOST:PORT/DBNAMEです。

接続先は環境変数DATABASE_URLから読み込む構成にします。本番の接続情報をalembic.iniへ固定せず、マイグレーションを実行する前に対象環境を確認します。

SQLAlchemyのモデルを定義する

エンジンとセッション

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

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}
    if DATABASE_URL.startswith("sqlite") else {},
)

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

アプリはSessionLocalからセッションを作り、処理後に閉じます。Alembicの設定は、この接続先と同じDATABASE_URLを参照するようにそろえます。

モデルと命名規約

# app/models.py
from datetime import datetime

from sqlalchemy import DateTime, MetaData, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=naming_convention)

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(
        String(200), nullable=False, index=True
    )
    description: Mapped[str | None] = mapped_column(String(1000))
    is_done: Mapped[bool] = mapped_column(default=False, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

命名規約は、主キー、外部キー、ユニーク制約、インデックスなどの名前を一定の形式で付けるルールです。名前が安定すると、生成された差分を比較しやすくなります。

Alembicで履歴管理を始めた後は、初回テーブルもBase.metadata.create_all()ではなくリビジョンから作成します。作成方法を一本化することで、空のデータベースにも同じ履歴を順番に適用できます。

Alembicを初期化する

設定ファイルを生成する

alembic init alembic

このコマンドでalembic/、alembic/env.py、alembic.iniが作成されます。

env.pyへメタデータを渡す

メタデータは、SQLAlchemyが保持するテーブル、カラム、制約などの定義です。target_metadataへBase.metadataを設定すると、Alembicがモデルと現在のデータベースを比較できるようになります。

# alembic/env.pyの主要部分
import os
from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool

from app.models import Base

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

target_metadata = Base.metadata

def get_url():
    url = os.getenv("DATABASE_URL")
    if url:
        return url
    return config.get_main_option("sqlalchemy.url")

def run_migrations_offline():
    context.configure(
        url=get_url(),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
        compare_server_default=True,
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(
        {"sqlalchemy.url": get_url()},
        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はサーバー側デフォルト値の違いを比較対象にします。自動検出の対象を増やしても、生成結果の確認は省略できません。

最初のリビジョンを作成して適用する

1. モデルと空のデータベースを比較し、リビジョンを生成します。

   alembic revision --autogenerate -m "create todos table"

2. alembic/versions/に作成されたファイルを開き、upgrade()とdowngrade()が意図した変更になっているか確認します。

3. 問題がなければ最新のリビジョンまで適用します。

   alembic upgrade head

4. 現在位置と履歴を確認します。

   alembic current
   alembic history --verbose

--autogenerateはレビュー前の下書きを作る機能として扱います。モデルから意図を完全に判断できるわけではないため、リビジョンファイルを読んでから適用します。

取り消しを試すときの注意

alembic downgrade -1
alembic upgrade head

downgradeは一つ前のリビジョンへ戻す操作ですが、削除したデータまで自動的に復元する保証にはなりません。ローカルや検証環境で往復を試し、本番ではバックアップと復旧手順も用意します。

既存データがあるテーブルを段階的に変える

NULLを許容してカラムを追加する

Todoに期限due_atを追加する例です。最初から非NULLにせず、既存行を更新できる状態を作ります。

# app/models.pyのTodoへ追加
due_at: Mapped[datetime | None] = mapped_column(
    DateTime(timezone=True)
)

alembic revision --autogenerate -m "add due_at to todos"
alembic upgrade head

非NULL化を3段階に分ける

1. 追加：NULLを許容したカラムを作る
2. データ更新：既存行の空欄を、要件に合う値で埋める
3. 制約変更：空欄が残っていないことを確認してから非NULLにする

カラム追加、データ更新、制約変更を別のリビジョンにすると、各段階で結果を確認できます。データ更新の値はアプリの要件によって異なるため、固定の値を機械的に入れず、対象データを確認して決めます。

SQLiteではBatchモードを使う

SQLiteでカラムの変更や削除を扱うときは、Alembicのbatch_alter_table()を使います。

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

revision = "xxxx_make_due_at_not_null"
down_revision = "xxxx_fill_due_at"

def upgrade():
    with op.batch_alter_table("todos") as batch:
        batch.alter_column(
            "due_at",
            existing_type=sa.DateTime(timezone=True),
            nullable=False,
        )

def downgrade():
    with op.batch_alter_table("todos") as batch:
        batch.alter_column(
            "due_at",
            existing_type=sa.DateTime(timezone=True),
            nullable=True,
        )

FastAPI側とマイグレーションの責任を分ける

FastAPI側はSessionLocalとTodoモデルを使って通常のCRUDを実装し、スキーマの作成と変更はAlembicの履歴に任せます。アプリ起動時のcreate_all()とAlembicを併用して変更経路を増やさないことが、状態を追いやすくするポイントです。

ローカルとCIでは、必要なリビジョンを適用した後にアプリのテストを実行します。モデルが期待する構造とデータベースの現在位置を、alembic currentで切り分けられます。

チームで履歴を保つ運用ルール

• 一つの変更を小さくする：カラム追加、データ更新、制約変更を分け、レビュー対象を明確にします。
• 生成ファイルをコードレビューする：モデルだけでなく、リビジョンのupgrade()とdowngrade()も確認します。
• 複数のheadを放置しない：並行ブランチで履歴が分かれたら、alembic headsで確認してマージリビジョンを作ります。
• 空のデータベースから試す：CIでalembic upgrade headを実行し、最初から最新まで履歴がつながるか確かめます。
• 適用前に戻し方を決める：バックアップ、リストア、対象環境の確認を実行手順へ含めます。

alembic heads
alembic merge -m "merge heads" <head1> <head2>

データ移行、トランザクション、CIまで含めた運用を検討する場合は、DBマイグレーションの実践的な運用ガイドも参考になります。

よくある症状と確認順

運用コマンド早見表

SQLiteからPostgreSQLを見据える場合

スキーマの履歴管理とデータベース製品の切り替えは、別の作業です。Alembicのリビジョンを整えても、既存データの移送や製品ごとの差の確認まで自動で完了するわけではありません。

モデルではDateTime(timezone=True)やString(n)など、元記事で示した型を使い、特定のデータベースだけに依存する変更はリビジョンを分けて明示します。PostgreSQLを採用する条件は、FastAPIでPostgreSQLを選ぶ判断基準で確認できます。

安全に進めるための確認リスト

1. SQLAlchemyのモデルを必要最小限だけ変更する
2. --autogenerateでリビジョンを作る
3. upgrade()とdowngrade()を読む
4. ローカルまたは検証環境で適用する
5. 必要に応じて取り消しと再適用を試す
6. アプリのテストを実行する
7. 対象環境、バックアップ、復旧手順を確認してから本番へ適用する

モデル、リビジョン、レビュー、適用、テストという順序を固定すると、変更の理由と現在位置を追いやすくなります。差分生成の結果に任せきらず、一つずつ確認できる大きさで履歴を積み重ねてください。