green snake
Photo by Pixabay on Pexels.com
目次

失敗しない!FastAPI×SQLAlchemy×Alembicで始めるデータベース移行(マイグレーション)完全ガイド


✅ 要約(最初に結論と全体像)

  • この記事でできること
    FastAPIアプリで使うデータベースのスキーマ変更(マイグレーション)を、安全・再現可能・チームフレンドリーに実施できるようになります。SQLAlchemyでモデルを定義し、Alembicで履歴管理・自動差分生成・ロールバックを行う、現場で通用する実装手順を具体的にご紹介します。
  • 主なステップ
    1. プロジェクト構成を整える → 2) SQLAlchemy ORMモデルを定義 → 3) Alembicを初期化 → 4) 自動差分でリビジョン作成 → 5) upgrade/downgrade → 6) 運用の落とし穴と回避策(命名規約・ブランチ解消・SQLite特有の注意点)
  • 想定DB
    開発はSQLiteで始め、のちにPostgreSQLMySQLへ移行しても動く設計をベースに解説します。
  • 効果
    • 「手作業のALTERで壊した…」を防止
    • 変更が履歴化され、チームで同じ状態をいつでも再現
    • CI/CDにも組み込みやすい安定フローを構築

🎯 誰が読んで得をするか(具体像)

  • 個人開発のわたし(学部3年・Web初挑戦)
    SQLiteでToDoアプリを作成中。テーブルに「期限」や「優先度」をあとから安全に追加したい。コマンドをコピペで動かし、変化の理由を視覚的(履歴)に把握したい方。
  • 小さなチームのわたし(受託開発・3名体制)
    仕様変更が週1で届く現場。DB変更が衝突しやすく、同僚のローカルで再現できない問題が多発。命名規約・履歴管理・ロールバックまで含めて、事故ゼロ運用にしたい方。
  • 将来スケールを見据えるわたし(スタートアップ)
    とりあえずSQLiteで始め、ユーザー増に合わせてPostgreSQLへ移行予定。移行に強い構成と、自動差分生成を軸にした堅実なフローを学びたい方。

1. 準備:最小プロジェクト構成とインストール

1.1 ディレクトリ構成(まずは形から整える)

fastapi-db/
├─ app/
│  ├─ main.py
│  ├─ db.py
│  ├─ models.py
│  └─ schemas.py
├─ alembic/           # Alembicが生成(init後)
├─ alembic.ini        # Alembic設定
└─ .env               # DATABASE_URLなど(任意)

1.2 必要パッケージ

python3 -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn sqlalchemy alembic pydantic "psycopg[binary]"  # PostgreSQLへ移行予定なら
# SQLiteだけなら psycopg は不要です

用語:マイグレーション=DBスキーマ(テーブル・カラム・制約など)の変更をプログラムとして履歴管理し、upgrade(適用)やdowngrade(取り消し)を安全に繰り返せる仕組み。

1.3 データベースURLの決め方(開発はSQLiteでOK)

  • 開発(ローカル):sqlite:///./app.db
  • 将来の本番(例:PostgreSQL):postgresql+psycopg://USER:PASSWORD@HOST:PORT/DBNAME

.env(任意)に入れておくと切り替えが楽です。

要点まとめ

  • まずはを作る → 依存をインストール
  • DB URLは環境変数で切替、最初はSQLiteでOK

2. SQLAlchemy ORMモデルを定義(2.xスタイルで未来志向)

2.1 db.py:エンジンとセッション(同期版)

Alembicは基本的に同期エンジンで動かすのが簡単です(アプリが非同期でもOK)。

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

2.2 models.py:Declarative Baseと命名規約

命名規約を決めておくと、自動生成の差分が安定します(とても大事です)。

# app/models.py
from datetime import datetime
from sqlalchemy import String, DateTime, func, MetaData
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
    )

補足:server_default=func.now()はサーバ側でのデフォルト値。SQLiteでもCURRENT_TIMESTAMP相当が入り、PostgreSQL移行時にも自然です。

2.3 初回テーブル作成はAlembicに任せる

直接 Base.metadata.create_all() は呼びません。
以降は**Alembicのupgrade**でテーブルを作るのが原則です(履歴を一本化するため)。

要点まとめ

  • SQLAlchemy 2.xの型付きORMで将来に強いモデル
  • 命名規約で自動差分のブレを抑える
  • 初回からAlembic主導でテーブル作成

3. Alembicを初期化し、アプリのメタデータを連携

3.1 初期化

alembic init alembic
  • alembic/alembic.ini が生成されます。

3.2 alembic.ini を編集(DB URL)

sqlalchemy.urlコメントのままにし、環境変数から読む方式にしておくのが安全です。
(URLをベタ書きすると、本番に誤適用のリスクが上がります)

3.3 alembic/env.py:アプリのmetadataを参照

target_metadata に**アプリのBase.metadata**を渡すと、--autogenerate が効きます。

# alembic/env.py(抜粋・必要部分のみ)
import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Alembicの設定取得
config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# ---- アプリのメタデータを読み込む ----
from app.models import Base  # これが重要!
target_metadata = Base.metadata

# ---- DB URLは環境変数やalembic.iniから ----
def get_url():
    url = os.getenv("DATABASE_URL")
    if url:
        return url
    return config.get_main_option("sqlalchemy.url")  # 必要なら.iniで指定

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

要点まとめ

  • target_metadata に自分のBase.metadataを渡す
  • compare_type/compare_server_default有効
  • URLは環境変数起点にしておく

4. 初回マイグレーションの作成と適用(自動差分)

4.1 初回リビジョンを作る

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

alembic/versions/xxxxxxxxxxxx_create_todos_table.py ができます。
ファイルを開いて、生成されたDDLが意図と一致しているか確認します(これが最重要)。

4.2 適用する(DBに反映)

alembic upgrade head
  • これでSQLiteにtodosテーブルが作成されます。
  • 以降、すべてのスキーマ変更はリビジョンとして履歴化します。

4.3 ロールバックも試す

alembic downgrade -1  # ひとつ前へ
alembic upgrade head  # 最新へ戻す

本番前に往復テストを必ず実施。壊れないことが自信になります♡

要点まとめ

  • --autogenerate魔法ではない。必ず差分を目視レビュー
  • upgrade/downgrade往復で安全性を確認

5. スキーマ変更の実例(よくある要件を丸ごと再現)

5.1 カラム追加:「期限(due_at)」を追加したい

models.py を編集(モデル→マイグレーションの順で必ず進めます)。

# 追記(models.py)
from sqlalchemy import DateTime

class Todo(Base):
    __tablename__ = "todos"
    # ... 既存は省略 ...
    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

リビジョン生成 → 確認 → 適用:

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

5.2 既存値へデフォルトを入れたい(非NULL化への橋渡し)

段階的に安全に進めます。

  1. NULL許容で追加(上の手順)
  2. 既存行に一括値を流し込みupgrade()でデータ更新)
  3. 非NULL制約に変更(別リビジョンで)

例:既存レコードへdue_atに暫定日付を入れる

# versions/xxxx_fill_due_at.py(手で作る or --autogenerate後に追記)
from alembic import op
import sqlalchemy as sa
from datetime import datetime, timedelta

revision = "xxxx_fill_due_at"
down_revision = "前のリビジョンID"

def upgrade():
    conn = op.get_bind()
    default_due = datetime.utcnow() + timedelta(days=7)
    conn.execute(sa.text("UPDATE todos SET due_at = :d WHERE due_at IS NULL"), {"d": default_due})

def downgrade():
    # 取り消し:NULLに戻す
    conn = op.get_bind()
    conn.execute(sa.text("UPDATE todos SET due_at = NULL"))

続けて非NULL化(SQLiteは制約変更が苦手 → 後述のbatchで対応)。

5.3 インデックス・ユニーク制約

タイトルにユニーク制約を付ける例:

# models.py
from sqlalchemy import UniqueConstraint

class Todo(Base):
    __tablename__ = "todos"
    # ...
    __table_args__ = (
        UniqueConstraint("title", name="uq_todos_title"),
    )

→ リビジョン作成&適用でOK。

5.4 SQLiteのALTER制限に配慮(batch_alter_table)

SQLiteはカラム削除/変更が苦手。AlembicのBatchモードを使います。

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

要点まとめ

  • モデル→リビジョン→適用の順序を厳守
  • データ移行が必要なら段階的に(NULL許容→データ投入→非NULL化)
  • SQLiteはbatch_alter_tableで柔軟対応

6. FastAPIアプリ本体への接続(動くサンプル)

6.1 schemas.py(Pydantic)

# app/schemas.py
from datetime import datetime
from pydantic import BaseModel

class TodoBase(BaseModel):
    title: str
    description: str | None = None
    is_done: bool = False
    due_at: datetime | None = None

class TodoCreate(TodoBase):
    pass

class Todo(TodoBase):
    id: int
    created_at: datetime

    class Config:
        from_attributes = True  # SQLAlchemy 2.x: orm_mode相当

6.2 main.py(同期セッションで最小CRUD)

# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from app.db import SessionLocal
from app.models import Todo
from app.schemas import Todo, TodoCreate
from typing import List

app = FastAPI(title="Todos API with Alembic")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/todos", response_model=Todo)
def create_todo(payload: TodoCreate, db: Session = Depends(get_db)):
    todo = Todo(**payload.dict())
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return todo

@app.get("/todos", response_model=List[Todo])
def list_todos(db: Session = Depends(get_db)):
    return db.query(Todo).order_by(Todo.id.desc()).all()

@app.get("/todos/{todo_id}", response_model=Todo)
def get_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(404, "Todo not found")
    return todo

@app.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, payload: TodoCreate, db: Session = Depends(get_db)):
    todo = db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(404, "Todo not found")
    for k, v in payload.dict().items():
        setattr(todo, k, v)
    db.commit()
    db.refresh(todo)
    return todo

@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(404, "Todo not found")
    db.delete(todo)
    db.commit()
    return {"ok": True}

ここでの肝はアプリ側はいつもどおり、DB構造の更新はAlembicの履歴で反映する点です。

要点まとめ

  • Pydanticのfrom_attributes=TrueでORM→スキーマ変換
  • アプリのCRUDはシンプルに、スキーマ変更はAlembicに任せる

7. チーム運用で効く実践Tips(失敗を減らす習慣)

7.1 命名規約は先に決める

  • 例:uq_<table>_<column>, fk_<table>_<column>_<ref_table>
  • これがあるだけで、自動差分が安定しレビューも短縮。

7.2 モデル編集は小さく・こまめに

  • 一度にたくさん変えると衝突レビュー困難
  • 「カラム追加」「データ移行」「制約強化」は分割して順に適用。

7.3 ブランチが分かれたら(複数ヘッド)

  • alembic heads で確認、増えていたらマージリビジョンを作成:
    alembic merge -m "merge heads" <head1> <head2>
  • マージ後は必ずupgrade/downgrade往復を確認。

7.4 テスト前に空DBから最新へ

  • CIで毎回:
    1. 空DB作成 → 2) alembic upgrade head → 3) アプリテスト
  • “最初から最新まで”が通ることを常に保証

7.5 破壊的変更は段階的

  • 例:非NULL化、型変更、外部キー導入
    1. フィールド追加(NULL許容)
    2. 埋める(データ移行)
    3. 制約強化(非NULL・FK)
  • 本番データを守る最善策です。

要点まとめ

  • 規約・小分け・マージ・CI再現・段階的移行
  • 「面倒」を先回りで潰すと、後の事故ゼロにつながります♡

8. よくあるハマりどころと回避策

症状 原因 対策
--autogenerate が差分を拾わない target_metadata 未設定 / インポート漏れ env.pyアプリのBase.metadataを参照。モデルが確実に読み込まれるようにimport順も確認
SQLiteでカラム削除・型変更が反映されない SQLiteのALTER制限 op.batch_alter_table() を使う。必要に応じて一時テーブル再作成が行われる
本番で誤ったDBに適用 URLのベタ書き・環境変数管理不備 alembic.iniにURLを固定しない。環境変数優先&CIで表示し確認
headsが複数に増える 並行作業で別々にリビジョン作成 alembic headsalembic merge統合、往復テスト
自動差分が毎回出る 命名規約不一致 / サーバデフォルト表記ずれ 命名規約導入、compare_server_default=True、DBごとの差異を明示的にそろえる

要点まとめ

  • 99%はenv.py命名規約batchで解決
  • 本番URLの扱いは常に慎重

9. サンプル:種データ(シーディング)を入れたい

注意:Alembicのupgrade()アプリ本質のデータ投入をしすぎると、差し戻しが難しくなることがあります。基本はスキーマだけに留め、最小限のマスタデータに限定するのが安全です。

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

revision = "xxxx_seed_master_tags"
down_revision = "xxxx_prev"

def upgrade():
    op.create_table(
        "tags",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("name", sa.String(50), nullable=False, unique=True),
    )
    conn = op.get_bind()
    conn.execute(sa.text("INSERT INTO tags (name) VALUES ('work'),('life'),('study')"))

def downgrade():
    op.drop_table("tags")

要点まとめ

  • シードはマスタ程度に留める
  • 大量データは別ジョブアプリコードで実施

10. 将来のDB移行(SQLite → PostgreSQL)を見据える設計

  • 型選択DateTime(timezone=True)String(n)など移植性の高い型を採用。
  • デフォルト値:アプリ側のdefaultではなく**server_default**を優先(DBが値を入れる)
  • トランザクション:マイグレーションは短く小さく、ロック時間を抑える
  • インデックス:後から個別リビジョンで追加し、影響範囲を分ける
  • 拡張(PostgreSQL固有機能)を使う場合は、専用リビジョンを分けて明示

要点まとめ

  • “移植性ファースト”な型・デフォルト
  • 変更は小さく段階的

11. 実務で役立つ運用コマンド早見表

# 変更作業の基本
alembic revision --autogenerate -m "explain the change"  # 差分生成
alembic upgrade head                                     # 最新適用
alembic downgrade -1                                     # ひとつ戻す
alembic current                                          # 今どこ?
alembic history --verbose                                # 履歴一覧
alembic heads                                            # HEADが複数あるか確認
alembic merge -m "merge heads" <head1> <head2>           # マージリビジョン

要点まとめ

  • 手が覚えるまで同じ手順を繰り返す
  • historycurrent状態確認を習慣化

12. セキュリティと品質(番外編だけど大切)

  • 権限:本番DBは最小権限のユーザーで。スキーマ変更用ユーザーを分けるのも有効。
  • バックアップupgrade前に自動バックアップdowngradeよりもリストアが最後の砦です。
  • レビュー:自動差分は必ず目視。外部キー・非NULL化は**本当に安全?**を全員で確認。
  • 監査:リビジョンファイルの差分もコードレビューへ。CIで生成&適用テストを回すと安心です。

要点まとめ

  • 最小権限・バックアップ・レビュー・CIの四点セットが安心運用の鍵

13. 仕上げ:一気通貫のミニワークフロー(おさらい)

  1. モデルを直すmodels.py):必要最小限の差分だけ
  2. 差分リビジョン生成alembic revision --autogenerate -m "..."
  3. レビュー:生成スクリプトのupgrade/downgrade必ず目視
  4. 適用alembic upgrade head(ローカル→ステージング→本番の順)
  5. 往復テストdowngrade -1upgrade head(壊れないか確認)
  6. CI/CD:空DB→upgrade head→アプリテストを毎回実行
  7. 運用:変更は小分け、衝突はmerge、SQLiteのALTERはbatchで対応

要点まとめ

  • “モデル→リビジョン→レビュー→適用”の黄金パターンを身体で覚える

14. まとめ(今日からマイグレーションを「怖くない」に♡)

  • AlembicはDB変更のタイムマシン。いつでも行き来でき、チーム全員が同じ景色を見られます。
  • 命名規約と自動差分でレビュー負荷を減らし、SQLite特有の制約batch_alter_tableで上手に回避。
  • 破壊的変更は段階的に進め、往復テストCIの再現性で品質を担保。
  • これらを守れば、開発初期のSQLiteから本番のRDBMSへスムーズに移行できます。わたしもはじめはドキドキでしたが、手順が定まれば安心してスキーマを進化させられますよ♡

付録A:開発向け.envの例(任意)

# SQLite
DATABASE_URL=sqlite:///./app.db
# PostgreSQL例(使うときにコメント解除)
# DATABASE_URL=postgresql+psycopg://myuser:mypassword@localhost:5432/mydb

付録B:FastAPI起動コマンド

uvicorn app.main:app --reload

おつかれさまでした。ここまで動かせれば、もうマイグレーションは怖くありません。次は複数テーブルのリレーションや、本番移行前のゼロダウンタイム戦略にも挑戦してみましょうね。わたしも全力で応援しています♡

投稿者 greeden

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です

日本語が含まれない投稿は無視されますのでご注意ください。(スパム対策)