SQLAlchemy 2.0 #6 クエリ応用 — 結合、集計、サブクエリ、大量処理

プログラミング言語 Python SQLAlchemy データベース
読了 5分

モデル、セッション、リレーションまでそろったので、今度はクエリの表現力を広げる番です。2.0 の良いところは、ここで新しい文法が出てこないことです。すべて select() の上に組み立てられ、Core でも ORM でも同じ形です。今回は実務で繰り返し書くことになるパターン集です。例は第 5 回の User、Address モデルを引き続き使います。

scalars と execute — 結果を受け取る 2 つの形 #

まず結果の受け取り方から整理しないと混乱します。

main.py 
from sqlalchemy import select

# エンティティ 1 つを取得するとき: scalars()
users = session.scalars(select(User).where(User.name.like("佐%"))).all()
# → [User, User, ...] オブジェクトのリスト

# 複数の値を取得するとき: execute()
rows = session.execute(select(User.name, User.email)).all()
# → [('佐藤', 'sato@example.com'), ...] Row のリスト
for name, email in rows:
    print(name, email)

ルールはひとつです。SELECT の対象がエンティティ 1 つなら scalars()、複数カラムやエンティティ + 集計の混在なら execute() です。execute() でエンティティ 1 つを取得すると (User,) のように要素 1 つの Row に包まれて返ってきます。初心者が最もよくつまずくポイントです。

単一行の取得には専用メソッドがあります。

main.py 
user = session.get(User, 1)                  # 主キー取得。なければ None
user = session.scalars(stmt).first()         # 先頭行または None
user = session.scalars(stmt).one()           # ちょうど 1 行でなければ例外

one() は「必ず 1 件のはず」の取得(ユニーク条件の検索)に使います。0 件でも 2 件でも例外になるので、データの異常を早期に発見できます。

条件の組み合わせ — and、or、in #

main.py 
from sqlalchemy import or_

stmt = select(User).where(
    User.name.like("佐%"),                    # where に並べる = AND
    User.id.in_([1, 2, 3]),
)

stmt = select(User).where(
    or_(User.name == "佐藤", User.name == "鈴木")
)

where() に条件を並べると AND で結ばれます。OR だけ or_() で包みます。in_() にはリストだけでなくサブクエリも渡せます(後述)。

結合 — relationship が ON 句を知っています #

main.py 
# ORM の結合: ON 句は relationship の宣言から推論
stmt = (
    select(User.name, Address.email_address)
    .join(User.addresses)
    .where(Address.email_address.like("%@work.com"))
)

# リレーション宣言がない、または曖昧なとき: ON 句を明示
stmt = select(User.name, Address.email_address).join(
    Address, User.id == Address.user_id
)

# LEFT OUTER JOIN: 住所のないユーザーも含める
stmt = select(User.name, Address.email_address).join(User.addresses, isouter=True)

join(User.addresses) のようにリレーション属性を渡せば ON 句を書く必要はありません。第 5 回の eager loading(selectinload)と混同しやすいのですが、役割が違います。join() は SQL の WHERE や SELECT で別のテーブルを使うためのもので、selectinload() はリレーション属性を先に埋めておくためのものです。「住所が work.com のユーザーを探したい」は join、「ユーザー一覧とそれぞれの住所を全部表示したい」は selectinload です。

集計 — group_by と label #

main.py 
from sqlalchemy import func

stmt = (
    select(User.name, func.count(Address.id).label("address_count"))
    .join(User.addresses, isouter=True)
    .group_by(User.id)
    .having(func.count(Address.id) >= 2)
    .order_by(func.count(Address.id).desc())
)
for row in session.execute(stmt):
    print(row.name, row.address_count)  # label のおかげで名前でアクセスできる
  • func.任意の名前() は、その名前の SQL 関数をそのまま呼び出します。func.count、func.sum、func.max はもちろん、DB 固有の関数も使えます。
  • label() を付けると、結果の Row からその名前で読めます。集計カラムには習慣として付けるのがよいです。
  • 集計結果に条件をかけるときは where ではなく having です。SQL のルールそのままです。

サブクエリと EXISTS #

「住所が 1 件でもあるユーザー」は 2 通りで書けます。

main.py 
# IN + サブクエリ
subq = select(Address.user_id)
stmt = select(User).where(User.id.in_(subq))

# EXISTS: 相関サブクエリ
from sqlalchemy import exists

stmt = select(User).where(
    exists().where(Address.user_id == User.id)
)

集計結果を結合に使うには、subquery() で名前付きの派生テーブルを作ります。

main.py 
addr_count = (
    select(Address.user_id, func.count(Address.id).label("cnt"))
    .group_by(Address.user_id)
    .subquery()
)
stmt = (
    select(User.name, addr_count.c.cnt)
    .join(addr_count, User.id == addr_count.c.user_id)
)

サブクエリのカラムは subq.c.カラム名 でアクセスします。Core の Table.c と同じインターフェースです。

ページネーション — LIMIT・OFFSET とその限界 #

main.py 
page, per_page = 3, 20
stmt = (
    select(User)
    .order_by(User.id)                 # 順序を固定しないとページが混ざります
    .limit(per_page)
    .offset((page - 1) * per_page)
)

覚えておくべきことが 2 つあります。第一に、order_by のないページネーションは未定義動作です。DB は順序を保証しないので、ページごとに行が重複したり抜けたりしえます。第二に、OFFSET は読み飛ばす行も一度は読みます。深いページ(OFFSET 100000)はその分遅くなるので、無限スクロール系には「最後に見た id より大きいもの」を条件にするキーセット(keyset)方式が向いています。

main.py 
# キーセットページネーション: 深さに関係なく一定の速度
stmt = select(User).where(User.id > last_seen_id).order_by(User.id).limit(per_page)

大量処理 — 単位作業を迂回する #

数万件を session.add() で入れると、変更追跡のコストがそのまま積み上がります。大量処理では ORM の便利さをあきらめて、式を直接実行するのが正解です。

main.py 
from sqlalchemy import insert, update

# 大量 INSERT: 辞書のリストで executemany
session.execute(
    insert(User),
    [{"name": f"user{i}", "email": f"user{i}@example.com"} for i in range(10_000)],
)

# 大量 UPDATE: 条件に合う全行を 1 文で
session.execute(
    update(User).where(User.name.like("test%")).values(name="整理済み")
)
session.commit()

この方式はオブジェクトを作らず、セッションの identity map も経由しません。だから速いのですが、すでにセッションにロードされているオブジェクトには変更が自動反映されないという代償があります。大量 UPDATE のあと同じセッションでその行をまた使うなら、コミット後に再取得するのが安全です。

まとめ #

  • エンティティ 1 つは scalars()、カラム混在は execute() で受け取ります。単一行は get()、first()、one() を用途別に使います。
  • 結合はリレーション属性を渡せば ON 句が推論されます。絞り込み用は join()、リレーション属性のロード用は selectinload() と役割が違います。
  • 集計は func + label + group_by、集計への条件は having です。サブクエリは in_()、exists()、subquery() で組み立てます。
  • ページネーションには order_by が必須で、深いページは OFFSET の代わりにキーセット方式を使います。
  • 大量 INSERT・UPDATE はセッションの変更追跡を迂回して式を直接実行します。ロード済みオブジェクトとの不一致にだけ注意します。
  • 最終回はスキーマ変更を管理する Alembic、非同期対応、実戦のプロジェクト構成を扱います。
X