1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@shinononononono

SQLAlchemy2.0のTIPS

1
Posted at

導入

SQLAlchemy 2.0を実務で使う際に役立つtipsをまとめた記事です。
SQLAlchemy 2.0は公式ドキュメントの情報量が非常に多い一方で、チュートリアルをひと通り終えただけでは実務でそのまま使えることは意外と少ないです。
また、ブログや技術記事でも業務で耐えうるような実践的なパターンを体系的に扱ったものは少なく、調べても答えにたどり着きにくいジャンルです。
この記事では、そういった実務でよく迷うポイントに絞って、具体的なコードとともに記載をします。

対象読者

SQLAlchemy 2.0のチュートリアルを一通り済ませた方、FastAPIなどのフレームワークでORMを使っていて、戻り値の型やクエリの書き方に迷ったことがある方を対象としています。
そのため、この記事を通してSQLAlchemy 2.0について知ってもらうことで、同様のモヤモヤを抱えている方の一助になればと考えています。

モデル定義

よくあるモデル定義

SQLAlchemy 2.0では、モデル定義の書き方が1.x系から大きく変わっています。
最も重要な変更点は Mapped[型] と mapped_column() の組み合わせによる型アノテーションベースの定義です。
これにより、カラムの型情報がPythonの型ヒントとして明示されるため、IDEの補完や mypy などの静的解析が効くようになります。
Mapped[int] のように記述することで「このカラムはNOT NULLのint型」であることを表します。一方、Mapped[int | None] とすれば nullable なカラムとして扱われます。
relationship() の back_populates は、双方向リレーションを張る際に対になる側のクラスの属性名を指定します。User.addresses と Address.user が互いを参照し合う形です。

from datetime import datetime

from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    addresses: Mapped[list["Address"]] = relationship(
        back_populates="user"
    )  # back_populatesには逆方向の属性名を指定します。


class Address(Base):
    __tablename__ = "addresses"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    postal_code: Mapped[str] = mapped_column(String(10))
    city: Mapped[str] = mapped_column(String(50))
    line1: Mapped[str] = mapped_column(String(255))
    user: Mapped[User] = relationship(back_populates="addresses")

モデル定義（多対多用中間テーブル）

多対多のリレーションは、中間テーブルを Table オブジェクトとして定義し、relationship() の secondary に渡すことで実現します。中間テーブル自体はORMモデルとして扱わず、あくまでSQLAlchemyが内部的に結合に使うテーブルとして機能します。back_populates は一対多の場合と同様に、双方向で必ず対になるよう両クラスに指定します。
参考: https://docs.sqlalchemy.org/en/20/orm/basic_relationships.html#many-to-many

from sqlalchemy import Column, ForeignKey, Table
from sqlalchemy.orm import Mapped, mapped_column, relationship


article_topics = Table(
    "article_topics",
    Base.metadata,
    Column("article_id", ForeignKey("articles.id"), primary_key=True),
    Column("topic_id", ForeignKey("topics.id"), primary_key=True),
)


class Article(Base):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    body: Mapped[str]
    topics: Mapped[list["Topic"]] = relationship(
        secondary=article_topics,
        back_populates="articles",
        lazy="selectin",
    )  # secondary に中間テーブルを指定することで多対多を実現


class Topic(Base):
    __tablename__ = "topics"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    articles: Mapped[list[Article]] = relationship(
        secondary=article_topics,
        back_populates="topics",
    )  # back_populatesは必ず対になるよう両クラスに指定する

CRUD

基本クエリ（insert）

session.add() でORMオブジェクトをセッションに追加し、session.commit() で確定します。
子テーブルへの登録など、同一トランザクション内でPKを参照したい場合は session.flush() を挟みます。
flush() はセッション内にある変更をDBに対してSQL（INSERT）として送信しPKを確定させますが、commit() はまだ呼ばれていないため、変更はDB上では確定していません。

pythonuser_payload = {
    "email": request_body["email"],
    "is_active": request_body.get("is_active", True),
}
new_user = User(**user_payload)
session.add(new_user)
session.flush()  # PKを確定させて子テーブルに流用する

address_payload = {
    "user_id": new_user.id,
    "postal_code": request_body["postal_code"],
    "city": request_body["city"],
    "line1": request_body["line1"],
}
session.add(Address(**address_payload))
session.commit()

基本クエリ（bulk insert）

複数件を一括登録する方法は主に2つあります。
session.add_all() はORMのsave-update cascadeが効くため、relationship() で紐づいた子オブジェクトも一緒にSessionに追加されます。親と子をまとめて登録したい場合はこちらが自然です。
session.execute(insert(Model), [...]) は単一テーブルへ直接INSERT文を発行するため、cascadeは効きません。relationshipの自動解決が不要な場合に使います。

# add_all: save-update cascadeが効く
# Userに紐づくAddressも一緒にSessionに追加される
users = [
    User(email="alice@example.com", addresses=[Address(city="Tokyo", ...)]),
    User(email="bob@example.com"),
]
session.add_all(users)
session.commit()

# session.execute: 単一テーブルへの直接INSERT
# cascadeは効かない
from sqlalchemy import insert

session.execute(
    insert(User),
    [
        {"email": "alice@example.com"},
        {"email": "bob@example.com"},
        {"email": "carol@example.com"},
    ],
)
session.commit()

基本クエリ（update）

update() にモデルを渡し、.where() で対象を絞り .values() で更新内容を指定します。.returning() を使うと、UPDATE後の値をそのまま取得できるため、更新後のデータをレスポンスに使いたい場合に便利です。

pythonfrom sqlalchemy import update

update_values = {
    "email": request_body["email"],
    "is_active": request_body["is_active"],
    "updated_at": utc_now,
}

stmt = (
    update(User)
    .where(User.id == target_user_id)
    .values(**update_values)
    .returning(User.id, User.email)
)
session.execute(stmt)
session.commit()

基本クエリ（delete）

delete() も同様に .where() で対象を絞って実行します。物理削除ではなく論理削除（is_deleted フラグの更新）を使うケースも多いですが、その場合は update() で代替します。

pythonfrom sqlalchemy import delete

stmt = delete(Address).where(Address.user_id == target_user_id)
session.execute(stmt)
session.commit()

基本クエリ（select）

select() を実行する際、結果の受け取り方にはいくつかの選択肢があります。
SQLAlchemy 2.0では session.execute() が汎用的な実行メソッドですが、ORMオブジェクトを取得する場合は session.scalars() を使うのが基本です。session.execute() はRowオブジェクトを返すのに対し、session.scalars() はORMオブジェクトを直接返してくれるためです。

カラムは select()内で絞ることもできますが、Rowオブジェクトが返ってきてしまうので、基本的にload_only()メソッドでカラムを絞ります。

from sqlalchemy.orm import load_only

# 複数件
users = session.scalars(
    select(User)
    .options(load_only(User.id, User.email))
    .where(User.is_active == True)
).all()

# 単件（0件・複数件なら例外）
user = session.scalars(
    select(User).options(load_only(User.id, User.email)).where(User.id == user_id)
).one()

# 単件（存在しない場合はNone。複数件なら例外）
user = session.scalars(
    select(User).options(load_only(User.id, User.email)).where(User.id == user_id)
).one_or_none()

集計など Row として受け取りたい場合は session.execute() を使います。

pythonfrom sqlalchemy import func

stmt = select(User.id, func.count(Order.id)).join(User.orders).group_by(User.id)
for user_id, order_count in session.execute(stmt):
    print(user_id, order_count)

リレーションシップ読み込みテクニック

リレーションシップの読み込みは、遅延読み込み（lazy loading）、即時読み込み（eager loading）、非読み込み（no loading）の3種類に分類されます。ここでは lazy と eager の2つをおさえます。
参考: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#relationship-loading-with-loader-options

lazy load

lazy loadとは、クエリから関連オブジェクトを最初に読み込まずに返されるオブジェクトを指します。特定のオブジェクトで指定されたコレクションや参照が初めてアクセスされると、追加のSELECT文が発行されます。
relationship() のデフォルトは lazy="select" で、lazy loadが有効になっています。基本的にあえてlazy loadを使うことはないですが、関連テーブルが巨大な場合などはeager loadでSQLが膨らむため、lazy loadの方が適しています。

pythonfrom sqlalchemy import select
from sqlalchemy.orm import load_only

# relationship()のデフォルト(lazy="select")の場合
stmt = (
    select(User)
    .options(load_only(User.id, User.email, User.is_active))
    .where(User.id == target_user_id)
)
user = session.scalars(stmt).one()

# addressesへ最初にアクセスした瞬間に追加SELECTが発行される
for address in user.addresses:
    print(address.city, address.line1)

eager load

eager loadとは、関連するコレクションやスカラー参照が事前に読み込まれた状態でクエリから返されるオブジェクトを指します。ORMはこれを実現するため、主クエリの後に追加のSELECT文を発行してコレクションを一括読み込みします（selectinload）。JOINで同時取得する joinedload という選択肢もありますが、コレクションの取得には selectinload が基本です。

selectinload を使うと、主クエリの後に IN 句を使った1本のSELECTで全ユーザーの addresses をまとめて取得するため、N+1問題を回避できます。

pythonfrom sqlalchemy import select
from sqlalchemy.orm import load_only, selectinload

stmt = (
    select(User)
    .options(
        load_only(User.id, User.email),
        selectinload(User.addresses).load_only(
            Address.id, Address.city, Address.line1
        ),
    )
    .where(User.is_active.is_(True))
)
users = session.scalars(stmt).all()

for user in users:
    print(user.email, "=>", [addr.city for addr in user.addresses])
1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?