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に戻れること」。抽象化に溺れず、道具としての自由度を味方につけましょう。