GitHubスター履歴

ライブラリ

SQLAlchemy

概要

SQLAlchemyは「The Database Toolkit for Python」として知られるPython最古で最も成熟したORMライブラリです。「SQL + Alchemy（錬金術）」の名の通り、多彩なデータベースのプログラミングを可能にし、SQL文の組み立てを自動化します。MySQL、PostgreSQL、SQLiteなど主要データベースエンジンに対応し、3つの抽象化レベルで柔軟な開発を支援します。

詳細

SQLAlchemy 2.0は2025年現在も活発に開発されている、Python界のデファクトスタンダードORMです。「Object-Relational Mapping」として、Pythonのクラス・オブジェクトとデータベースのテーブル・行を「つなぐ」仕組みを提供し、開発者がPythonコードだけでデータベース操作を完結できます。

主な特徴

• 3レベルの抽象化: ORM、SQL式言語、低レベルexecuteメソッドを組み合わせ可能
• 豊富なデータベース対応: MySQL、PostgreSQL、SQLite、Oracle、MS-SQL等をサポート
• セッション管理: 高度なトランザクション管理とオブジェクトライフサイクル
• リレーション機能: One-to-One、One-to-Many、Many-to-Manyの完全サポート
• マイグレーション: Alembicとの連携による強力なスキーマ管理
• 型ヒント対応: Python 3.6+の型ヒントと完全統合

メリット・デメリット

メリット

• Python界で最も歴史と実績のあるORM（20年以上の開発実績）
• 豊富なデータベースエンジンサポートによる高いポータビリティ
• 柔軟な抽象化レベル（生SQL〜フルORM）で段階的導入が可能
• 強力なクエリ構築機能とパフォーマンス最適化オプション
• FastAPIとの優れた相性で現代的なAPI開発に最適
• 充実したドキュメントと豊富なコミュニティリソース

デメリット

• 学習曲線が急でマスターに時間がかかる
• 設定が複雑で初心者には敷居が高い
• パフォーマンスチューニングに深い理解が必要
• ボイラープレートコードが多くなりがち
• 新しいORMと比較して開発者体験が劣る場合がある
• エラーメッセージが分かりにくいことがある

参考ページ

• SQLAlchemy 公式サイト
• SQLAlchemy 2.0 ドキュメント
• SQLAlchemy GitHub

書き方の例

基本的なセットアップ

pip install sqlalchemy
pip install sqlalchemy[postgresql]  # PostgreSQL用
pip install alembic  # マイグレーション用

モデルの定義（2.0スタイル）

from typing import List, Optional
from sqlalchemy import String, Integer, ForeignKey, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    email: Mapped[str] = mapped_column(String(100), unique=True)
    
    # One-to-Many関係
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", 
        cascade="all, delete-orphan"
    )
    
    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    
    # Many-to-One関係
    user: Mapped["User"] = relationship(back_populates="addresses")
    
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

データベース接続とセッション設定

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# データベース接続
engine = create_engine(
    "postgresql://username:password@localhost/dbname",
    echo=True,  # SQL出力をログ表示
    pool_size=10,
    max_overflow=20
)

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

# セッションファクトリ作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# セッション使用パターン
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

基本的なCRUD操作

from sqlalchemy import select

# セッション作成
with Session(engine) as session:
    # Create - ユーザー作成
    new_user = User(
        name="john_doe",
        fullname="John Doe",
        email="[email protected]",
        addresses=[
            Address(email_address="[email protected]"),
            Address(email_address="[email protected]")
        ]
    )
    session.add(new_user)
    session.commit()
    
    # Read - データ検索
    # 全ユーザー取得
    stmt = select(User)
    users = session.execute(stmt).scalars().all()
    
    # 条件付き検索
    stmt = select(User).where(User.name == "john_doe")
    user = session.execute(stmt).scalar_one_or_none()
    
    # 主キーで検索
    user = session.get(User, 1)
    
    # Update - データ更新
    stmt = select(User).where(User.name == "john_doe")
    user = session.execute(stmt).scalar_one()
    user.fullname = "John F. Doe"
    session.commit()
    
    # Delete - データ削除
    session.delete(user)
    session.commit()

高度なクエリ操作

from sqlalchemy import and_, or_, func, desc

with Session(engine) as session:
    # JOIN操作
    stmt = (
        select(User, Address)
        .join(User.addresses)
        .where(User.name == "john_doe")
    )
    results = session.execute(stmt).all()
    
    # 複雑な条件
    stmt = select(User).where(
        and_(
            User.name.like("john%"),
            or_(
                User.email.contains("gmail"),
                User.email.contains("yahoo")
            )
        )
    )
    users = session.execute(stmt).scalars().all()
    
    # 集計クエリ
    stmt = (
        select(User.name, func.count(Address.id).label("address_count"))
        .join(User.addresses)
        .group_by(User.name)
        .having(func.count(Address.id) > 1)
        .order_by(desc("address_count"))
    )
    results = session.execute(stmt).all()
    
    # サブクエリ
    subq = (
        select(func.count(Address.id))
        .where(Address.user_id == User.id)
    ).scalar_subquery()
    
    stmt = select(User).where(subq > 2)
    users = session.execute(stmt).scalars().all()

リレーションシップとEager Loading

from sqlalchemy.orm import selectinload, joinedload

with Session(engine) as session:
    # Lazy Loading（デフォルト）
    user = session.get(User, 1)
    addresses = user.addresses  # この時点でSQLが実行される
    
    # Eager Loading - selectinload
    stmt = select(User).options(selectinload(User.addresses))
    users = session.execute(stmt).scalars().all()
    
    # Eager Loading - joinedload
    stmt = select(User).options(joinedload(User.addresses))
    users = session.execute(stmt).unique().scalars().all()
    
    # 関係を通じたフィルタリング
    stmt = select(User).where(
        User.addresses.any(Address.email_address.like("%@gmail.com"))
    )
    gmail_users = session.execute(stmt).scalars().all()

トランザクション管理

# コンテキストマネージャー使用
with Session(engine) as session:
    try:
        user = User(name="test_user", email="[email protected]")
        session.add(user)
        
        address = Address(email_address="[email protected]", user=user)
        session.add(address)
        
        session.commit()
        print("トランザクション成功")
    except Exception as e:
        session.rollback()
        print(f"トランザクション失敗: {e}")

# 明示的なトランザクション制御
session = Session(engine)
transaction = session.begin()

try:
    # データ操作
    user = User(name="explicit_user", email="[email protected]")
    session.add(user)
    session.flush()  # INSERT実行（コミット前）
    
    # 何らかの処理
    user_id = user.id
    
    transaction.commit()
except Exception:
    transaction.rollback()
    raise
finally:
    session.close()

Many-to-Many関係の実装

from sqlalchemy import Table, Column

# 中間テーブル
student_course_table = Table(
    'student_course',
    Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
    Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True)
)

class Student(Base):
    __tablename__ = 'students'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    
    # Many-to-Many関係
    courses: Mapped[List["Course"]] = relationship(
        secondary=student_course_table,
        back_populates="students"
    )

class Course(Base):
    __tablename__ = 'courses'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    
    students: Mapped[List["Student"]] = relationship(
        secondary=student_course_table,
        back_populates="courses"
    )

# 使用例
with Session(engine) as session:
    student = Student(name="Alice")
    course1 = Course(title="Python Programming")
    course2 = Course(title="Data Science")
    
    student.courses.extend([course1, course2])
    session.add(student)
    session.commit()

生SQLの実行

from sqlalchemy import text

with Session(engine) as session:
    # 生SQL実行
    result = session.execute(
        text("SELECT name, email FROM user_account WHERE name = :name"),
        {"name": "john_doe"}
    )
    rows = result.fetchall()
    
    # DDL実行
    session.execute(text("CREATE INDEX idx_user_email ON user_account(email)"))
    session.commit()
    
    # ストアドプロシージャ実行（PostgreSQL例）
    result = session.execute(
        text("CALL update_user_stats(:user_id)"),
        {"user_id": 1}
    )

Alembicによるマイグレーション

# Alembic初期化
alembic init alembic

# マイグレーション作成
alembic revision --autogenerate -m "Create user and address tables"

# マイグレーション実行
alembic upgrade head

# マイグレーション履歴
alembic history

# 特定のバージョンにダウングレード
alembic downgrade -1

# alembic/env.py の設定例
from sqlalchemy import engine_from_config
from alembic import context
from myapp.models import Base

config = context.config
target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()