SQLAlchemy + Alembic 超入門

• 概要
• 環境
• インストール
• 初期化
• スキーマ定義
• 接続先DBの定義
• migrations/env.py の修正
• migration の実行
• テーブルの作成
• テーブル操作
• 最後に
• 参考サイト

概要

Python スクリプトでテーブル情報を定義しそのスキーマ情報から実際にテーブルを作成し操作するサンプルスクリプトを紹介します

環境

• macOS 15.7.1
• MySQL 9.4.0
• Python 3.12.11
  • mysqlclient 2.2.7
  • SQLAlchemy 2.0.43
  • alembic 1.16.5

インストール

• pipenv install sqlalchemy alembic mysqlclient

初期化

• pipenv run alembic init migrations

migrations/ と alembic.ini が作成されれば OK です

スキーマ定義

user テーブルを作成します
今回 SQLAlchemy は v2 なので mapped_column を使います

• mkdir lib
• touch lib/__ini__.py
• vim lib/models.py

from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    age: Mapped[int] = mapped_column(Integer, nullable=False)

接続先DBの定義

MySQL の接続先情報はここで定義します
必要に応じて環境変数化や設定ファイル化すれば環境ごとに接続先を変更できます

• vim lib/db.py

from sqlalchemy import create_engine

# MySQL で先に testdb というデータベースを作成しておくこと
# CREATE DATABASE testdb;
# MySQL ドライバは mysqlclient なので mysqldb を指定する
# 今回はテストなのでパスワード無しで root ユーザで接続
DATABASE_URL = "mysql+mysqldb://root:@localhost:3306/testdb"

engine = create_engine(DATABASE_URL, echo=True)


def get_engine():
    return engine

migrations/env.py の修正

alembic はデフォルトでは alembic.ini に記載されているデータベースに接続にいきます
今回は lib/db.py で接続先をコントロールするのでそれを使うように修正します

• vim migrations/env.py

from logging.config import fileConfig

from alembic import context

from lib.db import get_engine
from lib.models import Base

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline():
    """オフラインモード (SQL 出力)"""
    url = str(get_engine().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():
    """オンラインモード (DB へ実行)"""
    connectable = get_engine()

    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()

migration の実行

準備が整ったので migration していきます
まずは migration に必要なファイルを生成します

• pipenv run alembic revision --autogenerate -m "Create user table."

INFO  [sqlalchemy.engine.Engine] SELECT DATABASE()
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] SELECT @@sql_mode
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] SELECT @@lower_case_table_names
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [sqlalchemy.engine.Engine] BEGIN (implicit)
INFO  [sqlalchemy.engine.Engine] DESCRIBE `testdb`.`alembic_version`
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] DESCRIBE `testdb`.`alembic_version`
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] 
CREATE TABLE alembic_version (
        version_num VARCHAR(32) NOT NULL, 
        CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num)
)


INFO  [sqlalchemy.engine.Engine] [no key 0.00007s] ()
INFO  [sqlalchemy.engine.Engine] SHOW FULL TABLES FROM `testdb`
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [alembic.autogenerate.compare] Detected added table 'user'
INFO  [sqlalchemy.engine.Engine] ROLLBACK
  Generating /Users/user01/data/repo/python-try/migrations/versions/4c176b6b65d3_create_user_table.py ...  done

migrations/versions/4c176b6b65d3_create_user_table.py のようなファイルが作成できれば成功です

テーブルの作成

作成されたマイグレーションファイルから実際に MySQL にテーブルを作成します

• pipenv run alembic upgrade head

INFO  [sqlalchemy.engine.Engine] SELECT DATABASE()
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] SELECT @@sql_mode
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] SELECT @@lower_case_table_names
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [sqlalchemy.engine.Engine] BEGIN (implicit)
INFO  [sqlalchemy.engine.Engine] DESCRIBE `testdb`.`alembic_version`
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [sqlalchemy.engine.Engine] SELECT alembic_version.version_num 
FROM alembic_version
INFO  [sqlalchemy.engine.Engine] [generated in 0.00011s] ()
INFO  [sqlalchemy.engine.Engine] DESCRIBE `testdb`.`alembic_version`
INFO  [sqlalchemy.engine.Engine] [raw sql] ()
INFO  [alembic.runtime.migration] Running upgrade  -> 4c176b6b65d3, Create user table.
INFO  [sqlalchemy.engine.Engine] 
CREATE TABLE user (
        id INTEGER NOT NULL AUTO_INCREMENT, 
        name VARCHAR(255) NOT NULL, 
        age INTEGER NOT NULL, 
        PRIMARY KEY (id)
)


INFO  [sqlalchemy.engine.Engine] [no key 0.00006s] ()
INFO  [sqlalchemy.engine.Engine] INSERT INTO alembic_version (version_num) VALUES ('4c176b6b65d3')
INFO  [sqlalchemy.engine.Engine] [generated in 0.00014s] ()
INFO  [sqlalchemy.engine.Engine] COMMIT

実際に MySQL 側を確認するとテーブルができていることが確認できます

mysql> show create table user\G
*************************** 1. row ***************************
       Table: user
Create Table: CREATE TABLE `user` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL,
  `age` int NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
1 row in set (0.002 sec)

テーブル操作

簡単な操作方法を紹介します

• vim app.py

from lib.models import Base, User
from sqlalchemy import delete, select, update
from sqlalchemy.orm import Session

from lib.db import get_engine

# SQLite を例に使用
engine = get_engine()

# テーブル作成 (まだ存在しなければ)
Base.metadata.create_all(engine)

# セッション作成
with Session(engine) as session:
    # --- INSERT ---
    new_user = User(name="Alice", age=25)
    session.add(new_user)
    session.commit()
    print(f"Inserted User ID: {new_user.id}")

    # --- SELECT ---
    stmt = select(User).where(User.name == "Alice")
    result = session.scalars(stmt).all()
    for user in result:
        print(f"Selected: id={user.id}, name={user.name}, age={user.age}")

    # --- UPDATE ---
    stmt = update(User).where(User.name == "Alice").values(age=26)
    session.execute(stmt)
    session.commit()
    print("Updated Alice's age to 26")

    # --- DELETE ---
    stmt = delete(User).where(User.name == "Alice")
    session.execute(stmt)
    session.commit()
    print("Deleted Alice")

• pipenv run python app.py

で SQL が使えることを確認しましょう

最後に

SQLAlchemy v2 + alembic でテーブルマイグレーションを行う方法を紹介しました
テーブルなどが追加で必要であれば models.py に追記すれば OK です

参考サイト

• https://docs.sqlalchemy.org/en/20/orm/quickstart.html