SQL Alchemy実践入門：CoreとORMで“読めるSQL”を書く

SQLAlchemyとは、PythonでSQLを安全・表現力高く扱うためのORM兼SQLビルダーである。

なぜSQLAlchemyなのか（できることの全体像）

SQLAlchemyは「フレームワークに縛られず、SQLの素顔を保ったままPythonで扱える」ことが肝です。ORM（オブジェクト関係マッパー）とCore（SQLビルダー）の二層構造により、抽象度を自由にスライドできます。CRUDから複雑なCTEやウィンドウ関数、方言（Dialect）ごとの最適化まで、同じ思想で書けます。

代表的にできること

複数DB（SQLite/PostgreSQL/MySQL/Oracle/SQL Server等）を単一APIで横断
型安全なクエリ構築（プレースホルダと型情報でインジェクション耐性）
ORMでの宣言的モデリング、リレーション、ローダー戦略（N+1対策）
Coreでの生SQL級の制御（CTE、Window、ヒント、方言機能）
トランザクション/セッション管理（Unit of Work）
同期/非同期（AsyncEngine/AsyncSession）の両対応
マイグレーション（Alembic連携）

CoreとORMの違い（同階層比較）

観点	Core（SQLビルダー）	ORM（宣言的）
目的	SQLを構成要素で表現	テーブルをクラスとして扱う
粒度	低レベル・細粒度	高レベル・業務寄り
最適化	方言・ヒントまで自在	ローダー戦略＋必要に応じてCoreへ
学習コスト	SQL前提で素直	モデル/セッション理解が必要
ユースケース	レポート/集計/複雑サブクエリ	典型CRUD/ドメイン駆動

使ってみる：SQLiteで最小構成

ここではSQLiteで最小限のCRUDを通し、SQLAlchemy 2.0スタイル（select()中心）を掴みます。

1) インストールとエンジン作成

pip install "sqlalchemy>=2.0"

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("sqlite:///example.db", echo=False, future=True)

2) モデル定義（Declarative）

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, ForeignKey, select, func, Index

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), index=True, nullable=False)
    age: Mapped[int]
    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(100))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    author: Mapped["User"] = relationship(back_populates="posts")

# カスタムインデックスの例
Index("ix_posts_title", Post.title)

3) テーブル作成・データ投入・基本クエリ

Base.metadata.create_all(engine)

# セッションはUnit of Work（まとめて永続化を決める単位）
with Session(engine) as session:
    session.add_all([
        User(name="Alice", age=30),
        User(name="Bob", age=42),
    ])
    session.commit()

    # Insert + リレーション
    alice = session.execute(select(User).where(User.name == "Alice")).scalar_one()
    alice.posts.append(Post(title="はじめてのSQLAlchemy"))
    session.commit()

    # 取得：条件・並び替え・集約
    users = session.execute(
        select(User).where(User.age >= 30).order_by(User.age.desc())
    ).scalars().all()

    avg_age = session.execute(select(func.avg(User.age))).scalar_one()
    print(users, avg_age)

4) JOINとN+1回避（ローダー戦略）

from sqlalchemy.orm import joinedload, selectinload

with Session(engine) as session:
    # JOINしてまとめて取得（少数の親×少数の子に有効）
    res = session.execute(
        select(User).options(joinedload(User.posts)).order_by(User.id)
    ).scalars().all()

    # IN句で子を別クエリ取得（大量の親に有効）
    res2 = session.execute(
        select(User).options(selectinload(User.posts))
    ).scalars().all()

5) 更新・削除・トランザクション

from contextlib import contextmanager

@contextmanager

def transactional_session(engine):
    with Session(engine) as s:
        try:
            yield s
            s.commit()
        except:
            s.rollback()
            raise

with transactional_session(engine) as s:
    u = s.execute(select(User).where(User.name == "Bob")).scalar_one()
    u.age = 43
    s.delete(u)

モデリング：リレーション、制約、インデックス

リレーション：relationship(back_populates=...)で双方向、cascadeで親子の生存戦略を制御。
制約：nullable=False、UniqueConstraint、CheckConstraintで不変条件をDB側に表明。
インデックス：頻繁な検索列に付与（複合索引・部分索引はDB依存）。モデル定義時にIndexで宣言可能。
マイグレーション：Alembicのautogenerateで差分を作り、手で整えるのが定石。

クエリの書き方：読みやすさと最適化

2.0スタイルの原則：select(User)→.where(...)→.order_by(...)の直列合成。
CTE/Window：stmt = select(...).cte("name")、func.row_number().over(partition_by=..., order_by=...)で分析関数。
部分更新：update(User).where(...).values(age=User.age + 1)のようにCoreで一括更新。
バインド引数：where(User.name == bindparam("name"))でプレースホルダを明示。

ローダー戦略の比較（同階層比較）

戦略	仕組み	適性	罠
`joinedload`	親子をJOINで同時取得	少量の親×子	重複行で転送量増
`subqueryload`	子をサブクエリで取得	中規模	DBによって最適化差
`selectinload`	親IDのIN句で子を取得	親が多い	IN句が長大になる場合

トランザクションとセッション運用

短いスコープでwith Session(engine) as s:を基本に。スコープ外でオブジェクトに触れない（Expired/Detached対策）。
明示コミットとrollbackの徹底。コンテキストマネージャで雛形化すると安全。
分離レベル：create_engine(..., isolation_level=...)で要件に合わせる。
並列処理：スレッド/プロセスごとにセッションを分離。共有しない。

非同期・バルク処理・生SQLとの共存

非同期：create_async_engine(...)+AsyncSessionでI/O待ちに強いAPIを実現。
バルク：大量Insertはbulk_save_objectsやCoreのinsert()を検討（DBのCOPY系も視野に）。
生SQL：from sqlalchemy import textで部分的に併用可能。方言最適化の余地を残せる。

類似ライブラリとの違い（同階層比較）

ライブラリ	アーキテクチャ	特徴	向き/不得手
SQLAlchemy	Data Mapper＋Unit of Work＋Core	抽象度可変、方言最適化、非同期対応	大規模/複雑クエリ、最適化。初学コストは高め
Django ORM	Active Record	Djangoエコシステムと密結合、開発が速い	単体利用しづらい、複雑SQLは窮屈
Peewee	軽量ORM	シンプル・小規模向け	表現力・最適化は控えめ
Tortoise ORM	Async First（Active Record系）	非同期I/Oに親和、型ヒント良好	同期文脈や高度最適化は工夫が必要
SQLModel	SQLAlchemy + Pydantic	型駆動で宣言的。FastAPIと好相性	細かい最適化は結局SQLAlchemyへ

設計指針と落とし穴（中級者向けチェックリスト）

ドメインと永続化の分離：Repository/Service層を設け、ORMを境界に閉じ込める。
セッション境界の明示：リクエストごと/ジョブごとにスコープを切る。長寿命セッションはリークの温床。
N+1の定点観測：ローダー戦略を固定化せず、実行計画とログで随時点検。
インデックス設計：用途別に単一/複合/部分/ユニークを評価。不要なものは削除。
マイグレーション運用：自動生成→手修正→レビュー→本番適用のルーチン化。
型とスキーマ：Mapped[]型ヒントとnullable/unique/checkで“嘘をつけない”モデルに。
フェイルファスト：echo=TrueやSQLログで遅いクエリを即検知。

この入門のゴールは「SQLAlchemyで読めるSQLを書き、必要な時は素顔のSQLに戻れること」。抽象化に溺れず、道具としての自由度を味方につけましょう。