SQLAlchemy¹ を利用する場合に Alembic と sqlacodegen というツールを併用することで、コードベースでRDBのテーブルの設計や変更管理をすることが容易になりますので紹介します。他の著名なツールに例えると、Ruby on Rails における Active Record のマイグレーション機能のようなものになります。

論よりコードの方は、こちらに本記事で作成したコードを掲載しましたのでご覧ください。 github.com

なお、Alembic は Flaskの拡張や FastAPI でも活用されているツールであり、本記事はWebアプリの開発の文脈から切り離し、同ツール単体とRDBの変更管理にフォーカスしたものです。

概略
準備
フォワード操作
リバース操作
おまけ

概略

まず、SQLAlchemy を利用してRDBにCRUD操作をするユースケースがあったとします。主な構成要素と関係は次の通りとします。

上記ケースにおいて、Alembic を活用すると Code² から Table を生成する操作（Forward方向）ができ、sqlacodegen を活用すると Table から Code を生成する操作（Reverse方向）が可能となります。

したがって、python コードでRDBをハンドリングすることが可能となるため、git による管理のメリットを RDB の変更管理シーンにも持ち込む事ができますよ。という事が本記事でお伝えしたい内容になります。

準備

必要なパッケージをインストールします。alembic と sqlacodegen³ は一般的には開発者だけが利用すると思いますので dev dependencies に指定するとよいでしょう。

仮想環境作成＆パッケージインストール

$ python -V
Python 3.10.11
$ poetry init
...
$ poetry add "sqlalchemy^2.0" "psycopg2-binary^2.9"
$ poetry add alembic sqlacodegen_v2 --dev
$ poetry shell   # 仮想環境のセッションにはいっておく

session.py 作成

create_engine() 内の url には、実際に利用する DB への接続情報をセット⁴してください。

from contextlib import contextmanager

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


@contextmanager
def Session():
    engine = create_engine(
        url=f"postgresql+psycopg2://{dbusername}:{dbpassword}@{dbhost}:{dbport}/{dbname}",
        echo=True,
    )
    try:
        session = sessionmaker(bind=engine, autocommit=True, autoflush=True)
        with session() as sess:
            with sess.begin():
                yield sess
    finally:
        engine.dispose()

models.py 作成

DBのモデルを作成します。ここでは、SQLAlchemyのドキュメントに紹介されている UserとAddressモデルを例に利用します。

from datetime import datetime
from typing import List, Optional

from sqlalchemy import ForeignKey, Integer, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user"

    id = mapped_column(Integer, primary_key=True)
    name: Mapped[str]
    fullname: Mapped[Optional[str]]
    nickname: Mapped[Optional[str]] = mapped_column(String(64))
    create_date: Mapped[datetime] = mapped_column(insert_default=func.now())

    addresses: Mapped[List["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "address"

    id = mapped_column(Integer, primary_key=True)
    user_id = mapped_column(ForeignKey("user.id"))
    email_address: Mapped[str]

    user: Mapped["User"] = relationship(back_populates="addresses")

Alembic の環境を作る

Alembic チュートリアルの Creating an Environment に従ってAlembicの環境を新規生成します⁵。

$ alembic init --template generic ./alembic
  Creating directory '${workspaceFolder}/alembic/versions' ...  done
  Generating ${workspaceFolder}/alembic/script.py.mako ...  done
  Generating ${workspaceFolder}/alembic/env.py ...  done
  Generating ${workspaceFolder}/alembic.ini ...  done
  Generating ${workspaceFolder}/alembic/README ...  done
  Please edit configuration/connection/logging settings in '${workspaceFolder}/alembic.ini' before proceeding.
$

alembic.ini にDB接続情報をセットする

alembic.ini の sqlalchemy.url に接続情報をセットします。PostgreSQL へ psycopg2 を利用して接続する想定ですと以下のようになります。

# sqlalchemy.url = driver://user:pass@localhost/dbname
sqlalchemy.url = postgresql+psycopg2://testuser:veryS3cr3t@localhost:5432/testdb

alembic/env.py にモデルの基底クラスを指定する

モデルの配置先に合わせて次のように変更します。この変更は本記事で述べるモデルからリビジョンを自動生成する操作の動作条件です。Alembic の autogenerate 操作は、ワーキングディレクトリ配下にあるコードから、当基底クラスを継承したモデルを見つけて管理対象と認識します。

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

# target_metadata = None
from models import Base

target_metadata = Base.metadata   # モデルの基底クラスのメタデータを指定します

フォワード操作

準備が終わりましたので、コードからテーブルを生成してみましょう。

モデルからリビジョンを作成する

先に作成した models:User と models:Address モデルを元に、データベースへの変更内容を表現する リビジョン ファイルを生成します。また、この操作によってRDB上には内容が空の alembic_version テーブルも新規作成されます。つまり、リビジョンファイルとRDBのそれぞれが版の情報を持つことで、ローカルのコードとRDB上のテーブル間のリビジョンの一致・不一致がチェックされる仕組みです。

$ alembic revision --autogenerate -m "UserとAddressテーブルを追加したよ"
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'user'
INFO  [alembic.autogenerate.compare] Detected added table 'address'
  Generating ${workspaceFolder}/alembic/versions/879a5192b12c_userとaddressテーブルを追加したよ.py ...  done
$

リビジョンファイル⁶の中身を覗いてみると Revision ID と Revices の記載があります。それぞれ現リビジョンの番号と、前リビジョンの番号を示します。リビジョン作成操作が行われる度に、新たなリビジョンファイルが生成され、リビジョン間の前後関係、つまり変更履歴が保持される仕組みとなっています。これにより、あるリビジョンから遠く離れた特定のリビジョンの状態へ、戻したり進めたりするような操作⁷が可能となっています。

$ head alembic/versions/879a5192b12c_userとaddressテーブルを追加したよ.py
"""UserとAddressテーブルを追加したよ

Revision ID: 879a5192b12c
Revises: 
Create Date: 2023-07-12 21:46:23.294042

"""
from alembic import op
import sqlalchemy as sa

$

リビジョンをRDBに反映する

先ほど作成したリビジョンを反映する upgrade 操作をしてみましょう。パラメタに指定した head は最新リビジョンを反映する事を意味します。

$ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 879a5192b12c, UserとAddressテーブルを追加したよ
$

RDBをこの時点で確認すると、RDB上の alembic_version.version にはリビジョンがセットされ、User, Address テーブルがCreateされています。

リビジョンをあげる

さて、試しにUserモデルに enabled 列を追加する変更を加えてもう一度リビジョンを作成し、RDBに反映してみましょう。

モデルを変更し・・・

class User(Base):
    __tablename__ = "user"

    id = mapped_column(Integer, primary_key=True)
    name: Mapped[str]
    fullname: Mapped[Optional[str]]
    nickname: Mapped[Optional[str]] = mapped_column(String(64))
    create_date: Mapped[datetime] = mapped_column(insert_default=func.now())
    enabled: Mapped[Optional[bool]]  # 列を追加した

    addresses: Mapped[List["Address"]] = relationship(back_populates="user")

リビジョンを作り・・・

$ alembic revision --autogenerate -m "Userテーブルにenableカラムを追加したよ"
... 中略
  Generating ${workspaceFolder}/alembic/versions/d6f89f79414e_userテーブルにenableカラムを追加したよ.py ...  done
$ head alembic/versions/d6f89f79414e_userテーブルにenableカラムを追加したよ.py
"""Userテーブルにenableカラムを追加したよ

Revision ID: d6f89f79414e
Revises: 879a5192b12c
Create Date: 2023-07-12 22:10:22.806710

"""
from alembic import op
import sqlalchemy as sa

反映する。

$ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 879a5192b12c -> d6f89f79414e, Userテーブルにenableカラムを追加したよ
$

RDB上に User.enabled カラムの追加が反映されていれば成功です。

以上が、モデルからRDBへのフォワード操作となります。

リバース操作

翻って、RDBのテーブルからモデルを生成するリバース操作についてはどうでしょうか。

sqlacodegen

sqlacodegen_v2 というツールを利用する事ができます。ただし、SQLAlchemy ORMモデルは表記法が複数あるため、完全に意図した表記法での出力が得られるわけではない⁸点にご注意⁹ください。

リバース生成操作

reversed.py にモデルを出力させてみます。

## sqlacodegen_v2 --generator declarative ${DB接続文字列} > ${リバース生成コード出力先ファイル}
$ sqlacodegen_v2 --generator declarative postgresql+psycopg2://{dbusername}:{dbpassword}@{dbhost}:{dbport}/{dbname} > ./reversed.py
$

生成されたreversed.py

すると、以下のようなモデルコードが生成されます。内容を models.py へピックアップして編入してあげる用法となります。reversed.py 自体はただの出力結果ですので、見終わったら破棄して頂いて構いません。

from typing import List, Optional

from sqlalchemy import (
    Boolean,
    Column,
    DateTime,
    ForeignKeyConstraint,
    Integer,
    PrimaryKeyConstraint,
    String,
)
from sqlalchemy.orm import Mapped, declarative_base, mapped_column, relationship
from sqlalchemy.orm.base import Mapped

Base = declarative_base()


class User(Base):
    __tablename__ = "user"
    __table_args__ = (PrimaryKeyConstraint("id", name="user_pkey"),)

    id = mapped_column(Integer)
    name = mapped_column(String, nullable=False)
    create_date = mapped_column(DateTime, nullable=False)
    fullname = mapped_column(String)
    nickname = mapped_column(String(64))
    enabled = mapped_column(Boolean)

    address: Mapped[List["Address"]] = relationship(
        "Address", uselist=True, back_populates="user"
    )


class Address(Base):
    __tablename__ = "address"
    __table_args__ = (
        ForeignKeyConstraint(["user_id"], ["user.id"], name="address_user_id_fkey"),
        PrimaryKeyConstraint("id", name="address_pkey"),
    )

    id = mapped_column(Integer)
    email_address = mapped_column(String, nullable=False)
    user_id = mapped_column(Integer)

    user: Mapped[Optional["User"]] = relationship("User", back_populates="address")

reversed.py と元の models.py と比較するとかなりの表現差がありますので、正確な復旧性はなさそうだという事が分かります。復旧性はCreate文やバックアップの役割ですのでORMの役割に着目しますと、既存データベースのテーブルを初めてコード管理に編入したい場合に、ゼロからモデルをコーディングする手間をある程度削減する観点で、便利さを享受できるのではないかと思います。

おまけ

当記事で解説したリビジョン作成操作とマイグレーション操作を VSCode から起動する場合の launch.json は以下になります。upgrade操作を間違えて打ってしまわないようにだけご注意ください。

{
    "version": "0.2.0",
    "inputs": [
        {
            "id": "inputString",
            "type": "promptString",
            "description": "リビジョンコメントを入力してください",
        },
    ],
    "configurations": [
        {
            "name": "Alembic make rev",
            "type": "python",
            "request": "launch",
            "module": "alembic",
            "console": "integratedTerminal",
            "args": [
                "revision",
                "--autogenerate",
                "-m",
                "${input:inputString}",
            ],
        },
        {
            "name": "Alembic migrate to latest",
            "type": "python",
            "request": "launch",
            "module": "alembic",
            "console": "integratedTerminal",
            "args": [
                "upgrade",
                "head",
            ],
        },
    ]
}

以上となります。DB変更管理と変更オペレーション仕組み化のお役に立てば幸いです。

当記事の解説で利用する SQLAlchemy のバージョンは 2.0 を使用しています。前バージョンである 1.4 系とは構文が異なり互換性がありませんのでご注意ください。また、モデルのマッピングスタイルは執筆時点で利用者が多いと想定される宣言型マッピングを使用しています。宣言型マッピングについても2種類あり、Declarative Base スーパークラスを継承してモデルを作成する記法と、同スーパークラスの継承を用いずデコレータを利用するパターンの2種が選択可能ですが、本記事では前者を利用しています。↩
SQLAlchemy のマッピングモデルのことです。↩
パッケージ名は厳密には sqlacodegen のフォークである sqlacodegen-v2 です。フォーク元は SQLAlchemy 1.4 までしか対応しておらず、フォークは SQLAlchemy 2.0 スタイルに対応したものです。フォーク元のIssueを見ますと 2.0 での動作に関する言及もありましたので、将来的にはフォーク元を利用可能になるのではないかと思われます。↩
プロダクションコードでは、コードに接続情報をハードコーディングせず、環境変数経由でセットするかSecretsManagerなどから得た値をセットするプラクティスの使用をお薦めします。↩
蛇足ですが、複数DBや非同期アクセスの取り扱いに対応したAlembic環境をテンプレートから生成することもできます。今回はデフォルトである、単一DBと対象とした同期アクセスの取り扱いを想定した generic テンプレートを元に作成しています。↩
公式ドキュメントの表現に寄せてリビジョンファイルと表記していますが、マイグレーションスクリプトと呼ぶ方が通りが良いかもしれません。なぜなら、DjangoやFlask-Alembic、FastAPI などWebAppやAPIの開発において、モデルの変更内容をRDBへ反映する操作をマイグレーションと表現するためです。↩
ドキュメントの Relative Migration Identifier のセクションには、2世代新しい状態へのアップグレード upgrade +2, 1世代古い状態へのダウングレード downgrade -1, 特定のリビジョンの2世代後の状態へのアップグレード upgrade ae10+2 などの操作が紹介されています。↩
実際、100テーブルほどを対象にリバース生成したモデルから（本番とは異なるテスト用DBインスタンスへ）フォワーディングを試した際に、カラムのデータ型が変化したり制約が反映されなかったりと、大きな差が生じるケースもありました。↩
注意の仕方としては、リバース生成で得たモデルを初めてフォワード操作に用いる前には、リビジョンファイルに記述されているマイグレーション操作内容が意図したものであるかどうかをよく確認することと、事前のDBのバックアップを強くお勧めします。また、フォワード操作を担うAlembicは何ができて・何ができないのかにも触れておくと、注意すべき点をフォーカスしやすいと思います。↩

Alembic と sqlacodegen で RDB の変更管理をする

概略

準備