Python×SQLAlchemyで始めるORM入門: モデル定義からCRUDまで

python --version
python -c "import sqlite3; print('sqlite3 version:', sqlite3.sqlite_version)"

python -m pip install "SQLAlchemy>=2.0"

orm-tutorial/
├─ database.py     # EngineとBaseの定義
├─ models.py       # モデル定義(Userなど)
└─ crud_demo.py    # CRUDの実行スクリプト(実行例つき)

# database.py
# SQLAlchemy 2系スタイルでのBaseとEngine定義

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase

DB_URL = "sqlite:///./orm_tutorial.db"  # カレントディレクトリにDBファイルを作成

class Base(DeclarativeBase):
    """全モデルの共通基底クラス"""
    pass

# echo=Trueで発行SQLをログ出力(学習時は可視化が便利)
engine = create_engine(DB_URL, echo=True, future=True)

# models.py
# Userテーブルを表すモデルを定義(最小限のカラム構成)

from datetime import datetime
from sqlalchemy import String, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from database import Base  # 同一ディレクトリ想定

class User(Base):
    __tablename__ = "users"

    # 主キー。自動採番(INT AUTOINCREMENT相当)
    id: Mapped[int] = mapped_column(primary_key=True)

    # 必須項目。最大50文字
    name: Mapped[str] = mapped_column(String(50), nullable=False)

    # 一意制約とインデックスを付与
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)

    # 作成日時。Python側で挿入時に自動設定
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)

    def __repr__(self) -> str:
        return f"User(id={self.id}, name={self.name!r}, email={self.email!r})"

# 接続テストの例(任意)
# 既存のdatabase.pyのengineを流用して簡単な生SQLを実行
from database import engine

with engine.begin() as conn:
    result = conn.exec_driver_sql("SELECT 1")
    print("Connection OK:", result.scalar())

# init_db.py
# モデルに基づいてテーブルを作成するワンショットスクリプト

from database import engine
from models import Base

Base.metadata.create_all(bind=engine)
print("Tables created!")

# セッション基本形(例)
from sqlalchemy.orm import Session
from database import engine

with Session(engine) as session:
    # session.add(...) / session.execute(...) などを記述
    # 必要に応じてsession.commit()
    pass  # このブロックを出ると自動的にセッションが閉じます

# crud_demo.py
# 1ファイルでモデル定義からCRUD、例外処理までを実演します。

from __future__ import annotations

import os
from datetime import datetime

from sqlalchemy import String, DateTime, select, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

DB_FILE = "orm_tutorial.db"
DB_URL = f"sqlite:///./{DB_FILE}"

# 学習用に毎回クリーンな状態にする(既存DBを削除)
if os.path.exists(DB_FILE):
    os.remove(DB_FILE)

# Declarative Baseの用意
class Base(DeclarativeBase):
    pass

# モデル定義(Userテーブル)
class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)  # 主キー(INT, AUTOINCREMENT)
    name: Mapped[str] = mapped_column(String(50), nullable=False)  # 必須の文字列
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)  # 一意制約つき
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)  # デフォルト現在時刻

    def __repr__(self) -> str:
        return f"User(id={self.id}, name={self.name!r}, email={self.email!r})"

# Engine作成。学習ではecho=TrueでSQLログを見るのも有効ですが、出力を見やすくするためFalseにしています。
engine = create_engine(DB_URL, echo=False, future=True)

# テーブル作成
Base.metadata.create_all(bind=engine)

# セッションを開いてCRUDを順番に実行
with Session(engine) as session:
    # === Create: レコード追加 ===
    alice = User(name="Alice", email="alice@example.com")
    bob = User(name="Bob", email="bob@example.com")
    charlie = User(name="Charlie", email="charlie@example.com")

    # まとめて追加してコミット
    session.add_all([alice, bob, charlie])
    session.commit()  # ここでidが採番される

    print("After insert:", [alice, bob, charlie])

    # === Read: 全件と絞り込み ===
    all_users = session.scalars(select(User).order_by(User.id)).all()
    print("All users:", all_users)

    # 名前に'a'を含むユーザ( SQLiteのLIKEはASCIIでは大文字小文字を区別しないことが多い )
    users_with_a = session.scalars(
        select(User).where(User.name.like("%a%")).order_by(User.created_at.desc())
    ).all()
    print("Name contains 'a':", users_with_a)

    # 個別取得(getは主キー検索に最適)
    u2 = session.get(User, 2)
    print("Get id=2:", u2)

    # === Update: 値の変更とコミット ===
    u2.email = "bobby@example.com"  # 値を変更
    session.commit()                # 永続化
    session.refresh(u2)             # DB上の最新値でオブジェクトを更新
    print("Updated user id=2:", u2)

    # === Delete: 削除とコミット ===
    to_delete = session.get(User, 3)
    session.delete(to_delete)  # 削除を登録
    session.commit()           # 確定
    remaining = session.scalars(select(User).order_by(User.id)).all()
    print("Deleted id=3. Remaining:", remaining)

    # === 例外処理: 一意制約違反を誘発してロールバック ===
    try:
        # すでにalice@example.comが存在するため一意制約違反になる
        session.add(User(name="Eve", email="alice@example.com"))
        session.commit()
    except Exception as e:
        session.rollback()  # 直前のトランザクションを取り消す
        print("Rollback due to error:", type(e).__name__)

# 例外処理のテンプレート(部分例)
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.orm import Session
from database import engine
from models import User

with Session(engine) as session:
    try:
        session.add(User(name="Duplicate", email="alice@example.com"))
        session.commit()
    except IntegrityError as ie:
        session.rollback()
        print("Unique constraint violated:", ie.orig)  # DBドライバ由来の例外メッセージ
    except SQLAlchemyError as se:
        session.rollback()
        print("Database error:", se)
    except Exception as e:
        session.rollback()
        print("Unexpected error:", e)