DBマイグレーション（Alembic）¶

Alembic を使うと、テーブル定義を変えたときに、DBを作り直さずにスキーマを移行（マイグレーション）できる。

最低限のコマンド¶

Alembic をインストール済みで初期設定が終わっていれば、マイグレーションは基本的に次の2つの繰り返し。

# モデルを変更したら、マイグレーションファイルを生成
uv run alembic revision --autogenerate -m "変更の説明"

# 生成されたマイグレーションファイルを DB に適用
uv run alembic upgrade head

Note

プロジェクトで最初の1回だけ uv run alembic init alembic での初期化と設定が必要。
--autogenerate で生成したファイルは中身を確認して、必要なら手で調整する（cf. 実践）。

準備¶

まずは SQLModel のチュートリアルのとおりに DB のテーブルを作っておく。

# init_db.py
from sqlmodel import Field, SQLModel, create_engine


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

SQLModel.metadata.create_all(engine)

init_db.py を実行すると hero テーブルが作成される。

% sqlite3 database.db
SQLite version 3.51.0 2025-06-12 13:14:41
Enter ".help" for usage hints.
sqlite> .schema hero
CREATE TABLE hero (
    id INTEGER NOT NULL, 
    name VARCHAR NOT NULL, 
    secret_name VARCHAR NOT NULL, 
    age INTEGER, 
    PRIMARY KEY (id)
);

この DB を使って、 hero を READ・CREATE・DELETE する API を FastAPI で作成する。

# main.py
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
from sqlmodel import SQLModel, Field, create_engine, Session, select

app = FastAPI()


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None


sqlite_url = "sqlite:///database.db"
engine = create_engine(sqlite_url)


def get_db():
    with Session(engine) as session:
        yield session


@app.exception_handler(Exception)
async def internal_error_handler(request: Request, exc: Exception):
    print(f"Database error: {exc}")
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})


@app.get("/read-heroes")
def read_heroes(session: Session = Depends(get_db)):
    return session.exec(select(Hero)).all()


@app.post("/add-hero")
def add_hero(hero: Hero, session: Session = Depends(get_db)):
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero


@app.delete("/delete-hero")
def delete_hero(hero_id: int, session: Session = Depends(get_db)):
    hero = session.exec(select(Hero).where(Hero.id == hero_id)).first()
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"message": "The hero deleted successfully."}

Note

@app.exception_handler() を使って、データベース接続エラーが発生した時に呼び出される関数(internal_error_handler)を定義してる。
delete_hero 関数で raise してる status_code=404 のエラーは FastAPI がそのまま返すけど、データベース接続エラーとかほぼ全てのエラーが発生した時に internal_error_handler を呼び出してエラーハンドリングする。
cf. https://fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers

実践¶

ヒーローを何人か追加したあとに、例えば Hero クラスに email を足す。

# main.py: 抜粋
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(min_length=1, max_length=255)
    email: EmailStr = Field(min_length=1, max_length=255)
    secret_name: str
    age: int | None = None

テーブルが変わったので、 email 列がない既存テーブルに対してエラーになる。

Database error: (sqlite3.OperationalError) no such column: hero.email
[SQL: SELECT hero.id, hero.name, hero.email, hero.secret_name, hero.age 
FROM hero]
(Background on this error at: https://sqlalche.me/e/20/e3q8)

DB を削除して作り直せばいけるけど（データは消える・・・）、 Alembic を使えばデータを残したままマイグレーションできる。
まずはインストール。

uv add alembic

alembic init で初期化。

uv run alembic init alembic

Creating directory /Users/alice/code/alembic-test/alembic ...  done
Creating directory /Users/alice/code/alembic-test/alembic/versions ...  done
Generating /Users/alice/code/alembic-test/alembic/script.py.mako ...  done
Generating /Users/alice/code/alembic-test/alembic/env.py ...  done
Generating /Users/alice/code/alembic-test/alembic/README ...  done
Generating /Users/alice/code/alembic-test/alembic.ini ...  done
Please edit configuration/connection/logging settings in /Users/alice/code/alembic-test/alembic.ini before proceeding.

alembic.ini を編集して、接続先のDBを設定する。

# alembic.ini
#sqlalchemy.url = driver://user:pass@localhost/dbname  # この行はコメントアウト
sqlalchemy.url = sqlite:///database.db

alembic/env.py で、 target_metadata に SQLModel.metadata を設定する。

# alembic/env.py
from sqlmodel import SQLModel  # 追加
from main import Hero  # 追加

#target_metadata = None  # この行はコメントアウト
target_metadata = SQLModel.metadata

--autogenerate でマイグレーションファイルを作成する。モデルと実テーブルの差分（ email 列の追加）を検出してくれる。

uv run alembic revision --autogenerate -m "Add email to hero"

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare.tables] Detected added column 'hero.email'
  Generating /Users/alice/code/alembic-test/alembic/versions/2d990f0eccc3_add_email_to_hero.py ...  done

生成されたマイグレーションファイルの中身。

# alembic/versions/2d990f0eccc3_add_email_to_hero.py
"""Add email to hero

Revision ID: 2d990f0eccc3
Revises: 
Create Date: 2026-06-02 14:39:22.825020

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '2d990f0eccc3'
down_revision: Union[str, Sequence[str], None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('hero', sa.Column('email', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False))
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('hero', 'email')
    # ### end Alembic commands ###

生成されたままだと sqlmodel.sql.sqltypes を参照しているのに import がなくてエラーになるので、1行追加しておく。

# alembic/versions/2d990f0eccc3_add_email_to_hero.py: 抜粋
from alembic import op
import sqlalchemy as sa
import sqlmodel.sql.sqltypes  # 追加

マイグレーションを実行すると、今度は別のエラーになる。

uv run alembic upgrade head

(一部省略)
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) Cannot add a NOT NULL column with default value NULL
[SQL: ALTER TABLE hero ADD COLUMN email VARCHAR(255) NOT NULL]
(Background on this error at: https://sqlalche.me/e/20/e3q8)

email が必須（ NOT NULL ）として追加されるのに、既存のデータに email がないからエラーになっている。

Note

必須のカラムを追加するときは、一旦 nullable=True で追加 → 既存データを埋める → あらためて nullable=False にする、という2段階でマイグレーションするとよさそう。

まずはマイグレーションファイルを nullable=True に書き換える。

# alembic/versions/2d990f0eccc3_add_email_to_hero.py: 抜粋
    #op.add_column('hero', sa.Column('email', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False))
    op.add_column('hero', sa.Column('email', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True))

再度マイグレーションを実行。

uv run alembic upgrade head

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 2d990f0eccc3, Add email to hero

既存データの email を埋めておく。

% sqlite3 database.db
SQLite version 3.51.0 2025-06-12 13:14:41
Enter ".help" for usage hints.
sqlite> UPDATE hero SET email = 'temp@example.com' WHERE email IS NULL;

データが整ったら、あらためて email を非NULLにするマイグレーションファイルを作成する。

uv run alembic revision --autogenerate -m "Make email non-nullable"

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare.constraints] Detected NOT NULL on column 'hero.email'
  Generating /Users/alice/code/alembic-test/alembic/versions/fc628b8e5287_make_email_non_nullable.py ...  done

生成されたファイルの upgrade() を書き換える。 SQLite は ALTER COLUMN をサポートしていないので、 batch_alter_table を使う必要がある（ PostgreSQL などなら調整不要かも）。

# alembic/versions/fc628b8e5287_make_email_non_nullable.py: 抜粋
    #op.alter_column('hero', 'email', existing_type=sa.VARCHAR(length=255), nullable=False)
    with op.batch_alter_table('hero') as batch_op:
        batch_op.alter_column('email', existing_type=sa.String(length=255), nullable=False)

マイグレーションを実行。

uv run alembic upgrade head

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 2d990f0eccc3 -> fc628b8e5287, Make email non-nullable

これで email が必須の状態で FastAPI アプリを起動できる。

備考¶

上のやり方だと email は（DB的に）必須にできるけど、 EmailStr の形式チェックなどの Pydantic バリデーションは効いていない。 table=True のモデルは Pydantic のバリデーションがスキップされるから
本来ならモデル分離のように、入力用は非table のモデル（ HeroCreate ）に分けて定義しておくべき。そうすればバリデーションが効く。

cf. SQLModel のソースコード（sqlmodel_init の分岐）
cf. SQLModel Issue #453

class HeroBase(SQLModel):
    name: str = Field(min_length=1, max_length=255)
    email: EmailStr = Field(min_length=1, max_length=255)
    secret_name: str
    age: int | None = None


class HeroCreate(HeroBase):
    pass


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

add_hero 関数でも HeroCreate を受け取るようにする。

@app.post("/add-hero")
def add_hero(hero: HeroCreate, session: Session = Depends(get_db)):
    try:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero