Pythonでのマイグレーション導入検証（ローカル編）

はじめに

こんにちは！ポーラ・オルビスホールディングスで内製エンジニアをやっている高田です！
この記事では MySQL+SQLAlchemy+Python で構築した環境へのマイグレーションツールの導入検証をした結果を書いていきます。

• 背景 :
  • 現状、RDBMS としては AuroraRDS の MySQL 互換を、ORM として SQLAlchemy を採用している
  • バックエンドは Python でサーバレスの API を構築している
  • マイグレーションツールは未導入で、DB スキーマの修正を手作業で ALTER TABLE を実行している
• 目的 :
  • マイグレーション管理を導入したい
    • マイグレーションツールは SQLAlchemy と相性のいい Alembic を採用する想定
  • まずは SQLAlchemy + Alembic によるマイグレーションの動作をローカル環境で検証してみる
• 参考 : Alembic の公式ドキュメントは以下
  • https://alembic.sqlalchemy.org/en/latest/

最終的には AWS で管理している RDS のマイグレーションを実装したいのですが、ローカルで動かしてみないと課題が見えてこないのでまずはローカルで試してみようという趣旨の記事になります。

ファイル構成

• 最終的なファイル構成は以下の通り
• コマンド実行で自動生成されるやつもあるのでもろもろ後述します

test_local_migration/   # この検証を実施するためのディレクトリ
├── models/             # SQLAlchemyモデル定義
│   ├── __init__.py     # Alembicからmodels以下をまとめて読む込む
│   ├── base.py         # Base class、DB接続設定
│   ├── users.py        # Usersテーブル
│   ├── products.py     # Productsテーブル
│   └── orders.py       # Ordersテーブル
├── alembic/            # Alembicマイグレーションファイル（自動生成）
├── alembic.ini         # Alembicのiniファイル
├── .env                # 環境変数設定ファイル（MySQL用）
└── requirements.txt    # Python依存関係

• test_local_migration はワークディレクトリなので、各々好きな名前のフォルダにしていただければと思います

セットアップ手順

前提条件

• Python 3.8 以上がインストールされていること
  • 本検証は 3.12 で実施しました
• MySQL サーバーが起動していること
  • 今回はローカル環境に MySQL をインストールしています
• SQLAlchemy は導入済みであること

ステップ 1: 依存関係のインストール

まずは requirements.txt を以下のように記述します

alembic==1.13.2
PyMySQL==1.1.1
python-dotenv==1.0.0
SQLAlchemy==2.0.33

その後以下のコマンド実行

# 検証を実施するディレクトリに移動
cd test_local_migration

# 必要なPythonパッケージをインストール
pip install -r requirements.txt

• requirements.txt を使わない場合は以下の手順でも OK
  • pip install SQLAlchemy alembic で直接インストール
  • その後 pip freeze > requirements.txt で requirements.txt を出力して共有するなどの方法もあります
    • 上記のコマンドで requirements.txt を書き出すときは、環境が散らかるのを避けるため、適宜 venv 環境下などで実行するなどの工夫をしてくれればと思います

ここらへんはお好きなやり方で構築してくれればと

ステップ 2 : models 以下の実装

models 以下に SQLAlchemy を使って DB 接続するためのソースや、テーブルモデルファイルを作ります！

base.py の実装

SQLAlchemy を使って DB 接続するためのソースとして、base.pyを以下のように実装しておきます

base.py

import os
from urllib.parse import quote_plus


from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# .envファイルを読み込み
load_dotenv()

Base = declarative_base()


def get_database_url():
    """localの検証用のためハードコーディング"""
    db_host = "localhost"
    db_user = "root"
    db_password = "dummypw"
    db_name = "poc_database"
    db_port = "3306"

    # パスワードに特殊文字が含まれる場合はURLエンコードする
    if db_password:
        db_password = quote_plus(db_password)

    return f"mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"


def create_session():
    """データベースセッションを作成"""
    try:
        database_url = get_database_url()
        engine = create_engine(
            database_url,
            pool_pre_ping=True,  # 接続が生きているかチェック
            pool_recycle=3600,  # 1時間でコネクションを再作成
            echo=False,  # SQLログの出力（デバッグ時はTrueに）
        )
        SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
        return SessionLocal()
    except Exception as e:
        print(f"Database connection error: {e}")
        print(f"Database URL: {get_database_url()}")
        raise


def create_all_tables():
    """すべてのテーブルを作成（初期化用）"""
    engine = create_engine(get_database_url())
    Base.metadata.create_all(engine)

セッションを作成して DB 接続するためのソースって感じですね。
Base = declarative_base() のところに関してはマイグレーションツールとの連携の役割も持たせているので後のステップで説明します。

テーブルモデルファイルの実装

テーブルモデルファイルは SQLAlchemy の書き方に準じてテーブル定義を記載するだけです。
例を出すと以下のような記載になります。

users.py

from sqlalchemy import TIMESTAMP, Column, Integer, String, func
from .base import Base

class Users(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(TIMESTAMP, server_default=func.current_timestamp())

ここでは 1 テーブル 1 ファイルとなっているので、Alembic 側でまとめて読み込めるようにmodels/__init__.pyを以下のように定義します。

__init__.py

# models/__init__.py
from .base import Base
from .users import Users
from .products import Products
from .orders import Orders

__all__ = ['Base', 'Users', 'Products', 'Orders']

これをやっておかないとテーブルが読み込めず、逆にいまあるテーブルを消そうとしちゃうので忘れないようにしてください。（検証の中で一度この__init__.pyを定義し忘れてテーブルを焦土化しました。）

ステップ 3 : alembic の初期化

alembic init のコマンドを実行

以下コマンドを実行します。alembic init hogehugaみたいな感じでマイグレーション環境名は PJ に合わせて変えちゃってください。

# alembic init のあとはマイグレーション環境名
alembic init alembic

上記コマンド実行で以下が自動生成されます

test_local_migration/
├── alembic/            # Alembicマイグレーションファイル（自動生成）
└── alembic.ini         # Alembicのiniファイル（自動生成）

alembic.ini の編集

alembic.ini を開くと以下のような記載があるので

sqlalchemy.url = driver://user:pass@localhost/dbname

ここを以下のように実際の値に書き換える

sqlalchemy.url = mysql+pymysql://root:dummypw@localhost:3306/poc_database

• 実際はパスワードがハードコーディングだとまずいので、実装するときは Secrets Manager などから取得する仕組みにする必要があります
• 今回はいったんローカルの検証用 DB なのでハードコーディングで検証しちゃいました、パスワードも記事用の仮のものに置き換えてます

alembic/env.py の編集

alembic init alembicのコマンド実行で自動生成された alembic/ フォルダの中にalembic/env.pyというファイルがあります。
以下のように、alembic/env.pyのtarget_metadataを書き換えてあげる必要があります

# migrations/env.py の中で変更が必要な箇所
from models.base import Base  # ← 追加
import models  # ← これで models/__init__.py が実行され、全モデルが読み込まれる
target_metadata = Base.metadata  # ← 変更（初期はNoneになっているはず）

ステップ 3: 初回マイグレーション生成

以下のコマンドを実行します

# 現在のテーブルモデルから初回マイグレーションを生成
alembic revision --autogenerate -m "Initial migration"

すると以下のようなマイグレーションスクリプトが生成されます。

"""Initial migration

Revision ID: 61de41907eb2
Revises:
Create Date: 2025-07-10 15:04:42.108258

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


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


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('products',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.Column('name', sa.String(length=100), nullable=False),
    sa.Column('description', sa.Text(), nullable=True),
    sa.Column('price', sa.DECIMAL(precision=10, scale=2), nullable=False),
    sa.Column('stock', sa.Integer(), nullable=True),
    sa.Column('created_at', sa.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('users',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.Column('username', sa.String(length=50), nullable=False),
    sa.Column('email', sa.String(length=100), nullable=False),
    sa.Column('created_at', sa.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('orders',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.Column('user_id', sa.Integer(), nullable=False),
    sa.Column('product_id', sa.Integer(), nullable=False),
    sa.Column('quantity', sa.Integer(), nullable=False),
    sa.Column('order_date', sa.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=True),
    sa.ForeignKeyConstraint(['product_id'], ['products.id'], ),
    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
    sa.PrimaryKeyConstraint('id')
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('orders')
    op.drop_table('users')
    op.drop_table('products')
    # ### end Alembic commands ###

マイグレーション検証

次に実際にテーブルスキーマを修正してマイグレーションを実施して見ましょう

初回マイグレーション実行

テーブルのない空っぽの DB に対して、以下のコマンドを実行

$ alembic upgrade head

結果は以下の通り

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 61de41907eb2, Initial migration

その後 DB を確認すると、models 以下に定義した Users,Products,Orders の 3 テーブルが作成されたことが確認できました。

試しに users テーブルを参照すると以下のスキーマとなっていました

物理名	データ型	必須	デフォルト
id	int auto_increment	Yes
username	varchar(50)	Yes
email	varchar(100)	Yes
created_at	timestamp		CURRENT_TIMESTAMP

テーブルモデルファイルの実装の見出しで定義したスキーマと同一であることが確認できました

テーブルスキーマ修正

以下の通り、users テーブルに age を追加してみましょう

class Users(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), nullable=False)
    age = Column(Integer, nullable=True)  # ← 新しいカラムを追加
    created_at = Column(TIMESTAMP, server_default=func.current_timestamp())

    def __repr__(self):
        return (
            f"<Users(id={self.id}, username='{self.username}', email='{self.email}')>"
        )

その後マイグレーションスクリプトを以下のコマンドで生成

alembic revision --autogenerate -m "Add age column to users table"

versions の配下は以下のような形になります

alembic/
└── versions/
    ├── 61de41907eb2_initial_migration.py
    └── af30e8e64a19_add_age_column_to_users_table.py

新しくできたaf30e8e64a19_add_age_column_to_users_table.pyの中身を見ると以下の通り

"""Add age column to users table

Revision ID: af30e8e64a19
Revises: 61de41907eb2
Create Date: 2025-07-10 19:04:19.031907

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


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


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('users', 'age')
    # ### end Alembic commands ###

ここで alembic upgrade headコマンドを実行すると以下の通り、どこからどのバージョンにアップグレードされて、どんな変更を行ったかがコマンドラインから返されます

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 61de41907eb2 -> af30e8e64a19, Add age column to users table

実際のテーブル定義は以下の通り、age が追加されていました。

物理名	データ型	必須	デフォルト
id	int auto_increment	Yes
username	varchar(50)	Yes
email	varchar(100)	Yes
created_at	timestamp		CURRENT_TIMESTAMP
age	int

あとは以下のコマンドを実行すれば 1 つ前の age 追加適用前のバージョンに戻れます、逆もしかり
（head は最新のマイグレーションが適用されます。）

alembic downgrade -1

以上！ローカル検証終わり！