はじめに
どうも、水無月せきなです。
本記事は、シリーズ「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のインストールからマイグレーションの実行まで紹介しました。
次回は、アプリのアーキテクチャや具体的な実装について解説していきます。
ぜひ次回も合わせてお読みください！