SQLAlchemyとAlembicを活用したテーブル管理

そろそろPythonをちゃんとおぼえようかなと思いORMを触りはじめたのでメモがてら残します(来月には忘れていそうなので…) 。動的型付け言語が苦手というのもあって、正直なところあまり好みな言語ではない……かも。

目的

.NETのEntityFrameworkで便利な機能にEntityからマイグレーションスクリプトを生成してコードファーストでテーブルのバージョン管理ができるというものがあります。Pythonでも同じようなことができないかなと調べたところ、SQLAlchemy + Alembic で実現できそうだったので試してみました。

環境と準備

コンテナだったり仮想環境だったりの実行環境はすでに設定されていると想定します。 Python3.12.5とデータベースとしてPostgreSQL16を使います。

パッケージのインストール

SQLAlchemy
- ORM本体
Alembic
- SQLAlchemy と組み合わせて使うマイグレーションツール。SQLAlchemy のモデルクラスからマイグレーションスクリプトの生成とスクリプトのバージョン管理ができる
psycopg2
- PostgreSQLに接続するためのアダプタ

インストールする

pip install sqlalchemy alembic psycopg2-binary

初期化

インストールが完了すると alembic コマンドが使えるので、引数としてマイグレーションスクリプトを格納するディレクトリ名を指定して初期化をする(今回は alembic を指定)

alembic init alembic

ここまでで(大体)以下のようなディレクトリ構成になると思います

project_root/
│
├── alembic/
│   ├── versions/
│   ├── env.py
│   ├── README
│   └── script.py.mako
│
├── alembic.ini
└── main.py

初期設定

"alembic.ini" の sqlalchemy.url にデータベースへの接続情報を設定

sqlalchemy.url = postgresql://[username]:[password]@localhost/[dbname]

例えば

ユーザ名
- postgresuser
パスワード
- postgrespassword
データベース名
- db01

の場合は以下のようになります

sqlalchemy.url = postgresql://postgresuser:postgrespassword@localhost/db01

接続確認

接続の確認方法は少し悩みましたが、alemibicにデータベースに接続して現在のリビジョンを確認するコマンドがあるのでこれを使います。エラーが発生しなければ接続できていると思われます。

alembic current

ディレクトリ構成とか…

ここまでで準備ができたのでモデルクラスを作ります。ディレクトリ構成は自由ですが、今回は alembic ディレクトリと同じ階層に src ディレクトリを作りさらにその下の models ディレクトリに配置します。 (models の直下に __init__.py を作成しておきます)

project_root/
│
├── alembic/
│   ├── versions/
│   ├── env.py
│   ├── README
│   └── script.py.mako
│
├── alembic.ini
├── main.py
└── src/
    └── models/
        └── __init__.py

例として以下のテーブルを作成することを目的とします。

departments
- id
- department_code
- department_name
employees
- id
- employee_code
- employee_name
- department_code

モデルクラスを作る

今回は1ファイル1クラスの構成で、すべて models 内に作成します。まずモデルクラスの継承元となる DeclarativeBase を継承したクラスを作り、その後 Base クラスを継承した Department と Employee クラスを作ります。

base.py

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

depertment.py

from typing import TYPE_CHECKING

from typing import List, Optional
from sqlalchemy import String, Text, BigInteger
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .base import Base

if TYPE_CHECKING:
    from employee import Employee

class Department(Base):
    __tablename__ = "departments"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)

    department_code: Mapped[str] = mapped_column(String(5), unique=True, nullable=False)
    department_name: Mapped[str] = mapped_column(Text, nullable=False)

    employees: Mapped[List["Employee"]] = relationship("Employee", back_populates="department")

employee.py

from typing import TYPE_CHECKING
from sqlalchemy import String, Text, BigInteger, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .base import Base

if TYPE_CHECKING:
    from .department import Department

class Employee(Base):
    __tablename__ = "employees"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    employee_code: Mapped[str] = mapped_column(String(5), unique=True)
    employee_name: Mapped[str] = mapped_column(Text, nullable=False)
    department_code: Mapped[str] = mapped_column(String(5), ForeignKey("departments.department_code"))

    department: Mapped["Department"] = relationship("Department", back_populates="employees")

コードの説明(とinit.pyの追加)

__tablename__ にテーブル名を指定し、各項目にデータ型や制約を設定します。

指定できる型の一覧はこのあたりが参考になると思います…。 docs.sqlalchemy.org

あと if TYPE_CHECKING: の部分について。 DepartmentとEmployeeの2つのクラスは外部結合を表現するために循環参照をしていますがそのままお互いを循環importをするとマイグレーションスクリプト生成実行時にエラーが発生します。回避策として department.pyと employee.py それぞれからお互いをimportするのではなく、それぞれのクラスを使う予備もとでimportします(わかりづらいですが…)。今回は models/__init__.py に以下のように書きました。

from . import department
from . import employee

循環importの問題は解決できましたが、IDEやエディタに循環参照先の型を伝えるために以下の分岐があります。これにより型チェックの場合のみimportされます。(ここはIDEに型を伝えるためだけのコードなので、あっても無くても実行には影響ありません)。これは本当にどうかと思うのですが、今のところ他の方法がみつけられませんでした…。

if TYPE_CHECKING:
    from .department import Department

マイグレーションスクリプト生成

スクリプト生成の準備として alembic/env.py にマイグレーションの対象となるモデルクラスを追記します。今回の場合は以下のようになります。　

from src.models.base import Base
from src.models import department
from src.models import employee
target_metadata = Base.metadata

これで準備ができたのでマイグレーションスクリプトを生成します。文字列の部分は任意のメッセージを指定できます。

alembic revision --autogenerate -m  "create init table"

コマンドが成功すると alembic/versions ディレクトリにマイグレーションスクリプトが生成されます。

データベースに反映

最後に生成されたスクリプトをデータベースに反映します。

alembic upgrade head

成功するとデータベースにdepartments と employees テーブルの他に、マイグレーションのバージョン管理用のalembic_versionテーブルが作成されます。　

もうのお家(笑)

SQLAlchemyとAlembicでマイグレーションスクリプトを生成する

目的