はじめに

どうも、水無月せきなです。

本記事は、シリーズ「FastAPI × SQLModelで作るTodoアプリ」の第2回です。
前回の記事では、開発環境やプロジェクト構成について紹介しました。

今回は、データベースのマイグレーションを担うAlembicの導入と実際のマイグレーション方法について解説します。

シリーズ一覧

🛠️FastAPI × SQLModelで作るTodoアプリ①：開発環境とプロジェクトのセットアップ
🗄️FastAPI × SQLModelで作るTodoアプリ②：AlembicによるDBマイグレーション入門
📝FastAPI × SQLModelで作るTodoアプリ③：アーキテクチャと実装の詳細
🧪FastAPI × SQLModelで作るTodoアプリ④：テストの手法と実装

開発環境

Windows 11 24H2
WSL2
Docker Desktop 4.43.1
Cursor 1.2.2
Python 3.12
PostgreSQL 15

前提条件

SQLModelとデータベースドライバをインストール済みであること。
詳細は第１回をご参照ください。

Alembicのインストール

インストールするコマンド

uv add alembic

Alembicの初期化

app/migrationsの部分は、自分がマイグレーションファイルを管理したい場所に置き換えてください。

初期化のコマンド

uv run alembic init app/migrations

いくつかファイルが生成されるので、修正をしていきます。

alembic.iniの修正

alembic initで指定したパスを見るように、script_locationのパスを適宜修正します。
また、DBコンテナにアクセスできるように、接続文字列があっているか確認して適宜修正します。

※接続文字列は、後述するenv.pyで読み込む環境変数を元に組み立てられます。

alembic.ini

[alembic]

script_location = %(here)s/app/migrations

sqlalchemy.url = postgresql://%(POSTGRES_USER)s:%(POSTGRES_PASSWORD)s@%(POSTGRES_SERVER)s:%(POSTGRES_PORT)s/%(POSTGRES_DB)s

env.pyの修正

マイグレーション対象のモデルや、環境変数の設定などを追記・修正します。
ruffの警告が出るので、# ruff: noqaで抑制しました。

env.pyのコード

app/migrations/env.py

+ import os
from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool
+ from sqlmodel import SQLModel

+ # ruff: noqa
+ from app.models.todo import Todo

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
- target_metadata = None
+ target_metadata = SQLModel.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
+   config.set_section_option("alembic", "POSTGRES_USER", os.environ["POSTGRES_USER"])
+   config.set_section_option("alembic", "POSTGRES_PASSWORD", os.environ["POSTGRES_PASSWORD"])
+   config.set_section_option("alembic", "POSTGRES_DB", os.environ["POSTGRES_DB"])
+   config.set_section_option("alembic", "POSTGRES_SERVER", os.environ["POSTGRES_SERVER"])
+   config.set_section_option("alembic", "POSTGRES_PORT", os.environ["POSTGRES_PORT"])

    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)

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

script.py.makoの修正

マイグレーション時に動くスクリプトです（推測）。

sqlmodel.sql.sqltypesのインポートを追加します。
（FastAPIのサンプルプロジェクトを参照）

ruffやIDEのチェックが効かないので、大文字小文字を間違えてもそのまま通ってしまい、マイグレーションに失敗したりします（一敗）。

script.py.makoのコード

app/migrations/script.py.mako

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
+ import sqlmodel.sql.sqltypes
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}


def upgrade() -> None:
    """Upgrade schema."""
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    """Downgrade schema."""
    ${downgrades if downgrades else "pass"}

マイグレーションファイルを作成する

下記のコマンドを実行すると、app/migrations/versions 配下にマイグレーションファイルが生成されるはずです。

マイグレーションファイル生成

uv run alembic revision --autogenerate -m "first migration"

マイグレーションファイルを適用する

下記のコマンドを実行すると、DBへのマイグレーションが行われます。

マイグレーションを適用

uv run alembic upgrade head

補足

マイグレーションファイルの作成と適用は手動実行になるため、モデル定義の作成や変更がある度に作業が必要です。

おわりに

ここまでで、Alembicのインストールからマイグレーションの実行まで紹介しました。
次回は、アプリのアーキテクチャや具体的な実装について解説していきます。

ぜひ次回も合わせてお読みください！

参考資料