概要
過去にも紹介しましたがすでにテーブルがある際にわざわざテーブルのスキーマを定義しなくても Python からテーブルの情報を取得したりできる機能が automap です
今回は automap をすでに使っている環境で新規にテーブルを追加するスキーマを automap と共存させる方法を紹介します
同一 Metadata を使うのがポイントです
環境
- macOS 15.7.1
- MySQL 9.4.0
- Python 3.12.11
- mysqlclient 2.2.7
- SQLAlchemy 2.0.43
- alembic 1.16.5
テーブル準備
USE testdb;
CREATE TABLE user_legacy (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(100) NOT NULL,
email VARCHAR(255) NOT NULL UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
このテーブルは automap 用のテーブルになります
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
lib/models.py
モデルを定義するファイルです
automap の定義と追加するテーブルのスキーマ定義を共存させます
automap では既存のテーブルのみを管理するので reflection_options を追加します
reflection_options の設定がないと以下のエラーになります
sqlalchemy.exc.InvalidRequestError: Table 'user' is already defined for this MetaData instance. Specify 'extend_existing=True' to redefine options and columns on an existing Table object.
from sqlalchemy import Integer, MetaData, String
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from lib.db import get_engine
# 共通の metadata を作る
metadata = MetaData()
# Declarative 用 Base
class Base(DeclarativeBase):
metadata = metadata
# automap 用 Base
AutomapBase = automap_base(metadata=metadata)
# --- 既存DBをリフレクションしてクラス自動生成 ---
engine = get_engine()
# automap の対象となるテーブルを reflection_options で指定する
AutomapBase.prepare(
engine,
reflect=True,
reflection_options={"only": ["user_legacy"]}, # automap で読み込むテーブルを限定
)
# 例: 既存のテーブル user_legacy を ORM クラスとして取得できる
# UserLegacy = AutomapBase.classes.user_legacy
# --- 新規テーブルは DeclarativeBase で定義 ---
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)
マイグレーション確認
今回新規で追加するのは user テーブルになります
-
pipenv run alembic revision --autogenerate -m "Create user table."
これでマイグレーションファイルが作成できたら upgrade します
-
pipenv run alembic upgrade head
これで user テーブルが作成されていれば OK です
mysql> show tables;
+------------------+
| Tables_in_testdb |
+------------------+
| alembic_version |
| user |
| user_legacy |
+------------------+
3 rows in set (0.004 sec)
app.py で動作確認
automap と手動で定義したスキーマを使って各種 CRUD ができるか確認します
-
vim app.py
from sqlalchemy import delete, select, update
from sqlalchemy.orm import Session
from lib.db import get_engine
from lib.models import AutomapBase, Base, User
# SQLite / MySQL どちらでもOK
engine = get_engine()
# Declarative で定義したテーブルを作成 (まだ存在しなければ)
Base.metadata.create_all(engine)
# automap で読み込んだ既存テーブル
UserLegacy = AutomapBase.classes.user_legacy
# セッション作成
with Session(engine) as session:
# --- INSERT (user) ---
new_user = User(name="Alice", age=25)
session.add(new_user)
session.commit()
print(f"Inserted User ID: {new_user.id}")
# --- SELECT (user) ---
stmt = select(User).where(User.name == "Alice")
result = session.scalars(stmt).all()
for user in result:
print(f"Selected User: id={user.id}, name={user.name}, age={user.age}")
# --- UPDATE (user) ---
stmt = update(User).where(User.name == "Alice").values(age=26)
session.execute(stmt)
session.commit()
print("Updated Alice's age to 26")
# --- DELETE (user) ---
stmt = delete(User).where(User.name == "Alice")
session.execute(stmt)
session.commit()
print("Deleted Alice")
# -------------------------
# automap: user_legacy 操作
# -------------------------
# --- INSERT (user_legacy) ---
new_legacy = UserLegacy(username="bob", email="bob@example.com")
session.add(new_legacy)
session.commit()
print(f"Inserted UserLegacy ID: {new_legacy.id}")
# --- SELECT (user_legacy) ---
stmt = select(UserLegacy).where(UserLegacy.username == "bob")
result = session.scalars(stmt).all()
for ul in result:
print(f"Selected UserLegacy: id={ul.id}, username={ul.username}, email={ul.email}")
# --- UPDATE (user_legacy) ---
stmt = update(UserLegacy).where(UserLegacy.username == "bob").values(email="bob@newmail.com")
session.execute(stmt)
session.commit()
print("Updated bob's email")
# --- DELETE (user_legacy) ---
stmt = delete(UserLegacy).where(UserLegacy.username == "bob")
session.execute(stmt)
session.commit()
print("Deleted bob from user_legacy")
-
pipenv run python app.py
user_legacy も user もどちらも操作できることを確認しましょう
最後に
SQLAlchemy の automap と独自で定義したテーブルスキームを共存させる方法を紹介しました
automap はデフォルトだとすべてのテーブル情報を読み込んでしまうので reflection_options を使うのがポイントです
あとは Metadata を共有すれば OK です
すでにテーブルがあるサービスを automap で管理してる場合にどうしても新規でテーブルを追加しなければならないケースなどに使えるテクニックかなと思います
