失敗しない!FastAPI×SQLAlchemy×Alembicで始めるデータベース移行(マイグレーション)完全ガイド
✅ 要約(最初に結論と全体像)
- この記事でできること
FastAPIアプリで使うデータベースのスキーマ変更(マイグレーション)を、安全・再現可能・チームフレンドリーに実施できるようになります。SQLAlchemyでモデルを定義し、Alembicで履歴管理・自動差分生成・ロールバックを行う、現場で通用する実装手順を具体的にご紹介します。 - 主なステップ
- プロジェクト構成を整える → 2) SQLAlchemy ORMモデルを定義 → 3) Alembicを初期化 → 4) 自動差分でリビジョン作成 → 5)
upgrade
/downgrade
→ 6) 運用の落とし穴と回避策(命名規約・ブランチ解消・SQLite特有の注意点)
- プロジェクト構成を整える → 2) SQLAlchemy ORMモデルを定義 → 3) Alembicを初期化 → 4) 自動差分でリビジョン作成 → 5)
- 想定DB
開発はSQLiteで始め、のちにPostgreSQLやMySQLへ移行しても動く設計をベースに解説します。 - 効果
- 「手作業の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化への橋渡し)
段階的に安全に進めます。
- NULL許容で追加(上の手順)
- 既存行に一括値を流し込み(
upgrade()
でデータ更新) - 非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で毎回:
- 空DB作成 → 2)
alembic upgrade head
→ 3) アプリテスト
- 空DB作成 → 2)
- “最初から最新まで”が通ることを常に保証。
7.5 破壊的変更は段階的に
- 例:非NULL化、型変更、外部キー導入
- フィールド追加(NULL許容)
- 埋める(データ移行)
- 制約強化(非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 heads →alembic 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> # マージリビジョン
要点まとめ
- 手が覚えるまで同じ手順を繰り返す
history
とcurrent
で状態確認を習慣化
12. セキュリティと品質(番外編だけど大切)
- 権限:本番DBは最小権限のユーザーで。スキーマ変更用ユーザーを分けるのも有効。
- バックアップ:
upgrade
前に自動バックアップ。downgrade
よりもリストアが最後の砦です。 - レビュー:自動差分は必ず目視。外部キー・非NULL化は**本当に安全?**を全員で確認。
- 監査:リビジョンファイルの差分もコードレビューへ。CIで生成&適用テストを回すと安心です。
要点まとめ
- 最小権限・バックアップ・レビュー・CIの四点セットが安心運用の鍵
13. 仕上げ:一気通貫のミニワークフロー(おさらい)
- モデルを直す(
models.py
):必要最小限の差分だけ - 差分リビジョン生成:
alembic revision --autogenerate -m "..."
- レビュー:生成スクリプトの
upgrade
/downgrade
を必ず目視 - 適用:
alembic upgrade head
(ローカル→ステージング→本番の順) - 往復テスト:
downgrade -1
→upgrade head
(壊れないか確認) - CI/CD:空DB→
upgrade head
→アプリテストを毎回実行 - 運用:変更は小分け、衝突は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
おつかれさまでした。ここまで動かせれば、もうマイグレーションは怖くありません。次は複数テーブルのリレーションや、本番移行前のゼロダウンタイム戦略にも挑戦してみましょうね。わたしも全力で応援しています♡