SQLAlchemy 2.0の使い方｜CRUD・リレーションを動くコードで解説

SQLAlchemy 2.0とは、PythonのORM（オブジェクトとテーブルを対応づけて、クラス操作でSQLを組み立てる仕組み）であるSQLAlchemyの「型ヒントで定義し、select()で問い合わせる」新しい書き方の世代です。 2026年10月時点の最新版は 2.1.1（PyPI、2026年9月25日リリース）で、2.0の書き方がそのまま使えます。この記事のコードはすべて 2.0／2.1の両方に対応した書き方です。

検索上位の記事は、1.4から2.0への移行メモや、INSERTだけを扱った入門が中心でした。この記事では、モデル定義 → CRUD → リレーション → N+1の解消 → トランザクション → 一括INSERTの速度 → よくあるエラー → Alembic までを、SQLite と PostgreSQL 16（psycopg 3）の両方で実際に動かした出力と、echo=True で表示された発行SQL つきで解説します。コードはコピーしてそのまま動きます。

SQLAlchemy 2.0の早見表

動作環境とインストール

この記事の実行環境は次のとおりです。 バージョンはすべてPyPIで最新を確認して入れています。

# SQLiteだけなら sqlalchemy のみでOK（SQLiteはPython標準）
pip install "sqlalchemy>=2.0"

# PostgreSQLも使う場合（psycopg 3 のバイナリ版）
pip install "sqlalchemy>=2.0" "psycopg[binary]"

python -c "import sqlalchemy; print(sqlalchemy.__version__)"
# 2.1.1

2.1で知っておきたい変更は2つです。 公式の移行ガイド（Migration to 2.1）によると、対応Pythonは3.11以上になり、greenletが自動では入らなくなりました（asyncioを使う場合は pip install "sqlalchemy[asyncio]"）。同期のORMを使う範囲では、2.0のコードは2.1でもそのまま動きます（この記事のコードで確認済み）。

接続URLの書き方は「方言+ドライバ://ユーザー:パスワード@ホスト:ポート/DB名」です。

モデルを定義する（DeclarativeBase・Mapped・mapped_column）

2.0からは、型ヒント Mapped[型] で列を宣言します。 型から列の型とNULL可否が自動で決まるので、エディタの補完と型チェックが効きます。

ポイントは「Mapped[str] はNOT NULL、Mapped[str | None] はNULL許可」になることです。 mapped_column() を省略しても、型だけで列が作られます。

CRUDを1ファイルで書く（コピペ用の完成コード）

次の crud.py は、テーブル作成から作成・取得・更新・削除までを1ファイルにまとめたものです。 引数でDBのURLを渡せば、SQLiteでもPostgreSQLでも同じコードが動きます。

# crud.py  SQLAlchemy 2.0/2.1 ORM の CRUD を1ファイルで
import sys
from datetime import datetime
from sqlalchemy import create_engine, select, update, delete, func, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)
    nickname: Mapped[str | None]                      # NULL 許可（Optional）
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    tasks: Mapped[list["Task"]] = relationship(back_populates="owner", cascade="all, delete-orphan")

    def __repr__(self) -> str:
        return f"User(id={self.id}, name={self.name!r})"


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    done: Mapped[bool] = mapped_column(default=False)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="tasks")

    def __repr__(self) -> str:
        return f"Task(id={self.id}, title={self.title!r}, done={self.done})"


url = sys.argv[1] if len(sys.argv) > 1 else "sqlite:///app.db"
engine = create_engine(url, echo=True)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

# Create
with Session(engine) as session:
    alice = User(name="Alice", email="alice@example.com")
    alice.tasks = [Task(title="備品の発注"), Task(title="会議室の予約")]
    session.add_all([alice, User(name="Bob", email="bob@example.com", nickname="bobby")])
    session.commit()
    print("作成:", alice.id)

# Read
with Session(engine) as session:
    user = session.get(User, 1)                                   # 主キーで1件
    print("get:", user)
    stmt = select(User).where(User.name.like("A%")).order_by(User.id)
    print("scalars:", session.scalars(stmt).all())                # ORMオブジェクトのリスト
    row = session.execute(select(User.name, User.email).where(User.id == 2)).one()
    print("execute(列指定):", row, row.name)                     # Row（タプル風）
    print("件数:", session.scalar(select(func.count()).select_from(User)))

# Update
with Session(engine) as session:
    user = session.scalars(select(User).where(User.email == "bob@example.com")).one()
    user.nickname = "bob2"                                        # 属性を変えて commit
    session.commit()
    session.execute(update(Task).where(Task.owner_id == 1).values(done=True))  # 一括UPDATE
    session.commit()

# Delete
with Session(engine) as session:
    session.delete(session.get(User, 2))                          # 1件削除
    session.execute(delete(Task).where(Task.title == "会議室の予約"))  # 条件で削除
    session.commit()
    print("残り:", session.scalars(select(Task)).all())

python crud.py                                                       # SQLite
python crud.py "postgresql+psycopg://user:pass@localhost:5432/mydb"  # PostgreSQL

print の実行結果（SQLite・PostgreSQLとも同じ）

作成: 1
get: User(id=1, name='Alice')
scalars: [User(id=1, name='Alice')]
execute(列指定): ('Bob', 'bob@example.com') Bob
件数: 2
残り: [Task(id=1, title='備品の発注', done=True)]

echo=Trueで見える発行SQL（SQLite）

create_engine(url, echo=True) にすると、実際に発行されたSQLがログに出ます。 ORMが何をしているかを確かめる一番の近道です（ログの日時部分は省略しています）。

CREATE TABLE users (
    id INTEGER NOT NULL,
    name VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    nickname VARCHAR,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    PRIMARY KEY (id),
    UNIQUE (email)
)

BEGIN (implicit)
INSERT INTO users (name, email, nickname) VALUES (?, ?, ?) RETURNING id, created_at
[generated in 0.00061s (insertmanyvalues) 1/2 (ordered; batch not supported)] ('Alice', 'alice@example.com', None)
INSERT INTO users (name, email, nickname) VALUES (?, ?, ?) RETURNING id, created_at
[insertmanyvalues 2/2 (ordered; batch not supported)] ('Bob', 'bob@example.com', 'bobby')
INSERT INTO tasks (title, done, owner_id) VALUES (?, ?, ?) RETURNING id
...
COMMIT
BEGIN (implicit)
SELECT users.id, users.name, users.email, users.nickname, users.created_at
FROM users
WHERE users.id = ?
[generated in 0.00020s] (1,)
作成: 1

ここから読み取れることが3つあります。

• Mapped[str | None] の nickname だけが NOT NULLなし で作られている
• alice.tasks に入れたTaskは、owner_id を書かなくても 親のidが自動で入る（('備品の発注', 0, 1)）
• commit() の後に alice.id を読むと SELECTがもう一度発行される。commit時にオブジェクトの値が「期限切れ（expire）」になり、次に触ったとき読み直すためです（後述の expire_on_commit）

PostgreSQL（psycopg 3）では発行SQLがこう変わる

同じコードでもPostgreSQLでは、複数行のINSERTが1文にまとめられます。 SQLAlchemy 2.0から入った「insertmanyvalues」という仕組みで、RETURNINGつきでも一括で送れます。

CREATE TABLE users (
    id SERIAL NOT NULL,
    ...
    created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT now() NOT NULL,
    ...
)

INSERT INTO users (name, email, nickname) SELECT p0::VARCHAR, p1::VARCHAR, p2::VARCHAR FROM (VALUES (%(name__0)s::VARCHAR, %(email__0)s::VARCHAR, %(nickname__0)s::VARCHAR, 0), (%(name__1)s::VARCHAR, %(email__1)s::VARCHAR, %(nickname__1)s::VARCHAR, 1)) AS imp_sen(p0, p1, p2, sen_counter) ORDER BY sen_counter RETURNING users.id, users.created_at, users.id AS id__1
[generated in 0.00009s (insertmanyvalues) 1/1 (ordered)] {...}

UPDATE users SET nickname=%(nickname)s::VARCHAR WHERE users.id = %(users_id)s::INTEGER
DELETE FROM users WHERE users.id = %(id)s::INTEGER

session.get・scalars・execute の使い分け

迷ったら「ORMオブジェクトが欲しいなら scalars、列や集計なら execute」と覚えます。

結果の取り出し方は次のとおりです。

条件の書き方（where）逆引き

リレーション：1対多・多対多とN+1の解消

relationship() は、外部キーでつながったテーブルを属性としてたどれるようにする設定です。 両側に back_populates を書くと、片方を変えればもう片方も同期されます。

1対多・多対多のモデル（コピペ用）

# rel.py  1対多・多対多、N+1 の再現と解消
import sys
from sqlalchemy import create_engine, select, event, String, ForeignKey, Table, Column
from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column, relationship,
                            Session, selectinload, joinedload)


class Base(DeclarativeBase):
    pass


# 多対多の中間テーブル（記事 ⇔ タグ）
article_tag = Table(
    "article_tag", Base.metadata,
    Column("article_id", ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)


class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    articles: Mapped[list["Article"]] = relationship(back_populates="author")   # 1対多


class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Author] = relationship(back_populates="articles")            # 多対1
    tags: Mapped[list["Tag"]] = relationship(secondary=article_tag, back_populates="articles")


class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30), unique=True)
    articles: Mapped[list[Article]] = relationship(secondary=article_tag, back_populates="tags")


url = sys.argv[1] if len(sys.argv) > 1 else "sqlite://"
engine = create_engine(url)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

# 発行された SELECT の数を数える
count = {"n": 0}
@event.listens_for(engine, "before_cursor_execute")
def _count(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        count["n"] += 1

with Session(engine) as s:
    py, db = Tag(name="python"), Tag(name="db")
    for i in range(10):
        a = Author(name=f"著者{i}")
        a.articles = [Article(title=f"記事{i}-{j}", tags=[py, db] if j % 2 else [py]) for j in range(3)]
        s.add(a)
    s.commit()


def run(label, stmt):
    with Session(engine) as s:
        count["n"] = 0
        authors = s.scalars(stmt).unique().all()
        total = sum(len(a.articles) for a in authors)         # ここで遅延ロードが起きる
        tags = sum(len(art.tags) for a in authors for art in a.articles)
        print(f"{label:<28} 著者{len(authors)}人 記事{total}件 タグ{tags}個 → SELECT {count['n']}回")


run("遅延ロード（既定）", select(Author))
run("selectinload", select(Author).options(selectinload(Author.articles).selectinload(Article.tags)))
run("joinedload", select(Author).options(joinedload(Author.articles).joinedload(Article.tags)))
run("selectin+joined 混在", select(Author).options(selectinload(Author.articles).joinedload(Article.tags)))

# 多対多の逆方向
with Session(engine) as s:
    tag = s.scalars(select(Tag).where(Tag.name == "db").options(selectinload(Tag.articles))).one()
    print("tag=db の記事数:", len(tag.articles), sorted(a.title for a in tag.articles)[:3])

実行結果（SQLite・PostgreSQL 16 とも同じ回数）

遅延ロード（既定）                    著者10人 記事30件 タグ40個 → SELECT 41回
selectinload                 著者10人 記事30件 タグ40個 → SELECT 3回
joinedload                   著者10人 記事30件 タグ40個 → SELECT 1回
selectin+joined 混在           著者10人 記事30件 タグ40個 → SELECT 2回
tag=db の記事数: 10 ['記事0-1', '記事1-1', '記事2-1']

既定の遅延ロード（lazy="select"）では、著者1回＋著者ごとの記事10回＋記事ごとのタグ30回＝41回のSELECTが出ました。 これがN+1問題です。selectinload は「親のidを IN (...) にまとめて」関連ごとに1回ずつ取るので3回、joinedload はLEFT OUTER JOINで1回にまとめます。N+1の原因や他フレームワークでの見つけ方は別記事「N+1問題とは？」で扱っているので、ここではSQLAlchemyでの選び方だけ表にします。

1.x の書き方との対応表

1.x系の session.query() は2.1でも「レガシーAPI」として動きます（下の確認コードで 3 が返りました）。新しく書くコードは右列の書き方にそろえると、型ヒントとasyncioの両方で同じ書き方が使えます。

# 1.x の Query API（レガシー）も動く
with SessionLocal() as s:
    print(s.query(User).filter(User.email.like("%@example.com")).count())
# 3

移行で気をつけたい違いは「重複除去」です。 1.xの Query は、エンティティ1種類だけを取る場合などに重複した行を自動で1つにまとめていましたが、select() ではまとめません。joinedloadのコレクションでは .unique()、JOINで重複する場合は .distinct() を付けます。

トランザクションとbulk insertの速度

session.begin() でまとめてcommit・rollback

with SessionLocal.begin() as s: は、ブロックを抜けたらcommit、例外が出たらrollbackします。 下のコードでは2件目の追加でUNIQUE違反が起き、同じブロックの b@example.com も入りません。

# misc.py（抜粋）  前提: User(id, email unique) と Item(id, name, user_id) を定義済み
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import IntegrityError

SessionLocal = sessionmaker(engine)

with SessionLocal.begin() as s:          # 抜けるとき commit、例外なら rollback
    s.add(User(email="a@example.com"))
try:
    with SessionLocal.begin() as s:
        s.add(User(email="b@example.com"))
        s.add(User(email="a@example.com"))  # UNIQUE 違反
except IntegrityError as e:
    print("IntegrityError:", type(e.orig).__name__)
with SessionLocal() as s:
    print("users:", s.scalars(select(User.email)).all())   # b も入っていない

# SQLite
IntegrityError: IntegrityError
users: ['a@example.com']
# PostgreSQL 16（psycopg）
IntegrityError: UniqueViolation
users: ['a@example.com']

e.orig にはドライバ側の元の例外が入ります。 PostgreSQLでは UniqueViolation のように種類まで分かるので、ログに残すと原因が追いやすくなります。

1万件INSERTの速度比較（実測）

大量INSERTは「ORMオブジェクトを作らず、辞書のリストを insert() に渡す」と速くなります。

N = 10_000
uid = 1
# ① 1件ずつ add + flush
for i in range(N):
    s.add(Item(name=f"i{i}", user_id=uid)); s.flush()
# ② add_all で一括
s.add_all([Item(name=f"i{i}", user_id=uid) for i in range(N)])
# ③ ORM の bulk INSERT（2.0 以降の推奨）
s.execute(insert(Item), [{"name": f"i{i}", "user_id": uid} for i in range(N)])
# ④ Core（Connection に直接）
s.connection().execute(Item.__table__.insert(), [{"name": f"i{i}", "user_id": uid} for i in range(N)])

PostgreSQLでは①に比べて③が約36倍速くなりました（同一コンテナ内で各1回の計測。環境で値は変わります）。①は1件ごとにDBと往復するため、ネットワーク越しのDBほど差が開きます。主キーなどを返してほしい場合は insert(Item).returning(Item.id) と組み合わせると、2.0以降はまとめて取得できます。

よくあるエラーと直し方（実行して再現）

DetachedInstanceError（セッションを閉じた後に関連を読んだ）

with SessionLocal() as s:
    u = s.scalars(select(User)).first()
print(u.items)                       # セッションを閉じた後に遅延ロード

DetachedInstanceError : Parent instance <User at 0x7f9465c736f0> is not bound to a Session; lazy load operation of attribute 'items' cannot proceed (Background on this error at: https://sqlalche.me/e/21/bhk3)

直し方：セッションの中で先に読み込んでおきます。

with SessionLocal() as s:
    u = s.scalars(select(User).options(joinedload(User.items))).unique().first()
print("joinedload なら閉じた後も読める:", u.items)
# joinedload なら閉じた後も読める: []

commit後に属性を読むとDetachedInstanceError（expire_on_commit）

s = SessionLocal()
u = User(email="c@example.com"); s.add(u); s.commit(); s.close()
print(u.email)

DetachedInstanceError : Instance <User at 0x7f9465bb6f90> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/21/bhk3)

commitで値が期限切れになり、閉じた後に読み直せないためです。 戻り値として使いたい場合は sessionmaker(engine, expire_on_commit=False) にすると、commit時点の値を保持します。

s = SessionLocal(expire_on_commit=False)
u = User(email="d@example.com"); s.add(u); s.commit(); s.close()
print("expire_on_commit=False:", u.id, u.email)
# expire_on_commit=False: 3 d@example.com

joinedloadでunique()を付け忘れる

InvalidRequestError : The unique() method must be invoked on this Result, as it contains results that include joined eager loads against collections

session.scalars(stmt).unique().all() のように .unique() を挟みます。 コレクションの場合は selectinload に替えると .unique() は不要です。

one() で0件

NoResultFound : No row was found when one was required
one_or_none(): None

「無いこともある」検索は .one_or_none() か .first() を使います。

エラー早見表

Alembicでマイグレーションする（概要とコマンド）

Alembicは、SQLAlchemyのモデルの変更をDBに反映するマイグレーションツールです。 create_all() は「無いテーブルを作る」だけで列の追加・変更はしないため、運用中のDBはAlembicで管理します。

pip install alembic
alembic init migrations                     # 雛形を作る
# alembic.ini の sqlalchemy.url を設定
# migrations/env.py で target_metadata = Base.metadata にする
alembic revision --autogenerate -m "create members"
alembic upgrade head
alembic current

INFO  [alembic.autogenerate.compare.tables] Detected added table 'members'
Generating .../migrations/versions/002fa7b1847c_create_members.py ...  done
INFO  [alembic.runtime.migration] Running upgrade  -> 002fa7b1847c, create members
002fa7b1847c (head)

autogenerateの結果は必ず目で確認します。 列名の変更は「削除＋追加」と判定されるため、データを残したい場合は op.alter_column() に書き換えます。

よくある質問

Q. SQLAlchemy 2.0と2.1はどちらを使えばよいですか？ A. 新しく始めるなら最新の2.1系で問題ありません。この記事のコードは2.0と2.1の両方で同じように動く書き方です。Python 3.10以前の環境では2.1が入らないので、2.0系を使います。

Q. session.query() はもう使えないのですか？ A. 2.1でもレガシーAPIとして動きます。ただし公式ドキュメントは select() を中心に書かれているため、新しいコードは select() にそろえると調べものがしやすくなります。

Q. FastAPIと組み合わせるときのセッションはどう作りますか？ A. SessionLocal = sessionmaker(engine) を作り、依存関数で with SessionLocal() as s: yield s とする形が基本です。レスポンスでモデルを返すなら、expire_on_commit=False かeager loadで必要な値を先に読み込んでおくとDetachedInstanceErrorを避けられます。

Q. 生のSQLを書きたいときは？ A. from sqlalchemy import text を使い、session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1}) のようにバインドパラメータで値を渡します。

まとめ

• 最新は2.1.1。2.0の書き方（DeclarativeBase・Mapped・mapped_column）がそのまま使える
• 取得は select() ＋ session.scalars()、列や集計は session.execute() / session.scalar()
• echo=True で発行SQLを確認すると、SQLiteとPostgreSQLの違いやcommit後の読み直しが見える
• コレクションは selectinload、多対1は joinedload で、41回のSELECTが3回・1回に減った
• 大量INSERTは session.execute(insert(Model), [dict])。PostgreSQLで1件ずつより約36倍速かった
• DetachedInstanceErrorはeager loadか expire_on_commit=False で解消
• スキーマ変更はAlembic、autogenerateの結果は目で確認する

株式会社RJCでは、SQLAlchemyのようなORMの使い方を仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。