--- title: "SQLModelとは？使い方・モデル定義・unique制約・FastAPI連携【0.0.48対応】" url: "https://www.issoh.co.jp/tech/details/10322/" published: 2025-12-18 updated: 2026-10-28 categories: ["Python"] publisher: "株式会社一創" --- # SQLModelとは？使い方・モデル定義・unique制約・FastAPI連携【0.0.48対応】 SQLModelは、SQLAlchemyのテーブル定義とPydanticのデータ検証を1つのクラスで兼ねる、Python用のORMです。FastAPIと同じ作者（Sebastián Ramírez）が開発しています。この記事では、SQLModelが何を解決するのか、SQLAlchemyとどう選び分けるのかを整理したうえで、モデル定義、ユニーク制約、CRUD、リレーション、FastAPI連携までを順に扱います。 コードはすべて、SQLModel 0.0.48（2026年10月6日リリース）、SQLAlchemy 2.1.6、Pydantic 2.14.1、FastAPI 0.162.0、Python 3.12.13の組み合わせで実行し、出力されたDDLやエラーメッセージをそのまま載せています。 ## まとめ：SQLModelの位置づけと使うときの要点 - SQLModelのクラスはSQLAlchemyのモデルであり、同時にPydanticのモデルでもあります。テーブル定義とAPIの入出力モデルを二重に書かずに済みます。 - 2026年10月時点の最新は0.0.48で、1.0未満です。0.0.45では`datetime`列の既定がUTC前提に変わるなど、0.0.x同士でも破壊的変更が入るため、版は固定して上げるたびにリリースノートを読みます。 - 単一列のユニーク制約は`Field(unique=True)`、複数列の組み合わせは`__table_args__`に`UniqueConstraint`を置きます。重複はcommit時の`IntegrityError`で捕まえます。 - クエリは`session.exec(select(...))`で書きます。`session.query()`は動きますが、SQLModelの`Session`では`DeprecationWarning`が出ます。 - `table=True`のモデルは直接生成しても検証が走りません。外部入力は検証用のモデルで受け、`model_validate`で変換します。 - FastAPIでは`lifespan`でテーブルを作り、セッションは依存性注入でリクエストごとに開きます。`async def`の中で同期の`Session`を使うと、DB待ちの間ほかのリクエストが止まります。 ## SQLModelとは：SQLAlchemyとPydanticを1クラスで兼ねるORM 公式サイトは、SQLModelを「Pydantic and SQLAlchemyの上に載る薄い層（a thin layer on top of Pydantic and SQLAlchemy）」と説明しています（[SQLModel公式ドキュメント](https://sqlmodel.tiangolo.com/)）。DBとのやり取りはSQLAlchemyのエンジンとセッションがそのまま担い、SQLModelはそこにPydanticの型検証と、1クラスで両方を表す仕組みを足しています。ORMそのものの仕組みは[ORMとは？O/Rマッパーの仕組み・メリットとデメリット](/tech/details/3755/)で解説しています。 ### 二重定義の解消：テーブルとAPI入出力モデルのフィールド共有 SQLAlchemyだけでAPIを作ると、テーブル用の`DeclarativeBase`のクラスと、リクエストやレスポンスを検証するPydanticの`BaseModel`を別々に書き、同じ「名前・年齢」を2か所で宣言して相互に変換するコードが要ります。SQLModelでは、`SQLModel`を継承したクラスの型ヒントがそのままカラム型にもPydanticの検証ルールにもなります。`table=True`を付けたクラスがテーブル、付けないクラスは検証専用のモデルです。共通の項目を基底クラスに置いて両者で継承すれば、フィールドの宣言は1回で済みます（後述のFastAPI章の`HeroBase`がこの形です）。 FastAPIは必須ではありません。SQLModel単体でもスクリプトやバッチのORMとして使えます。ただし、入出力モデルとテーブルを継承で共有できる利点はAPIを作るときに最も効くので、FastAPIとの連携は代表的な利用例です。Pydantic側の書き方は[Pydanticとは？BaseModelの使い方とV1との違い](/tech/details/4838/)が参考になります。 ### 開発状況：0.0.48・Python 3.10以上・依存するSQLAlchemyの範囲 [PyPIのsqlmodel](https://pypi.org/project/sqlmodel/)に登録されている最新版は0.0.48（2026年10月6日）で、要件はPython 3.10以上、依存は`SQLAlchemy>=2.0.14,<2.2.0`と`pydantic>=2.11.0`です。0.0.48で「SQLAlchemy 2.1のサポート」が加わりました。ただしSQLAlchemy 2.1系はPython 3.11以上が必要なので、Python 3.10で`pip install sqlmodel`すると2.0系が入ります。同じSQLModel 0.0.48でも、Pythonの版によって下のSQLAlchemyが変わる点は覚えておいてください。 直近の版には、既存コードの動きを変える変更が含まれています。 | 版 | リリース日 | 主な変更 | | ------ | ---------- | ------------------------------------ | | 0.0.45 | 2026-09-21 | `datetime`列の既定をUTC（タイムゾーン付き）に変更 | | 0.0.47 | 2026-09-23 | PostgreSQL・MySQL・MariaDBをCIのテスト対象に追加 | | 0.0.48 | 2026-10-06 | SQLAlchemy 2.1に対応 | 0.0.45の変更は[リリースノート](https://github.com/fastapi/sqlmodel/releases/tag/0.0.45)で「Breaking Changes」と明記されています。1.0に達していないライブラリは、マイナー相当の変更が0.0.xの末尾で入ります。本番では`sqlmodel==0.0.48`のように版を固定し、上げる前に[公式のリリースノート](https://sqlmodel.tiangolo.com/release-notes/)を読む運用が前提です。 ### SQLAlchemyとの違いと選び方：採用を避けたほうがよい場面 SQLModelはSQLAlchemyの上に載る層なので、SQLModelで書けないことはSQLAlchemyの書き方で補えます。違いは、どこまでを1つのクラスに寄せるかです。 | 観点 | SQLModel | SQLAlchemy単体 | | ------- | ---------------- | ------------------------ | | 入力の検証 | Pydanticで標準装備 | Pydantic等を別に組む | | モデル定義 | 型ヒント＋`Field` | `Mapped`＋`mapped_column` | | API用モデル | 継承で共有 | 別クラス＋変換 | | 版の段階 | 0.0.48（1.0未満） | 2.1.6 | | 非同期 | `AsyncSession`あり | `AsyncSession`あり | FastAPIでCRUD中心のAPIを作り、テーブルと入出力モデルの重複を減らしたいならSQLModelを選びます。逆に、次の条件に当てはまるならSQLAlchemyを直接使うほうが安全です。 - 版が上がるたびに回帰テストを回す体制が無いまま、長期運用の基幹システムに入れる場合（0.0.45のように0.0.x同士で既定値が変わるため）。 - 継承マッピング、複雑なサブクエリ、ハイブリッド属性など、SQLAlchemyの高度な機能が設計の中心にある場合。結局SQLAlchemyの書き方に降りることになり、1クラス化の利点が残りません。 - DB層だけを独立したパッケージとして提供し、Pydanticによるデータ検証や入出力モデルとのフィールド共有を必要としない場合。 SQLAlchemy自体の基本は[SQLAlchemyとは？読み方・使い方とPython ORMの基本](/tech/details/7912/)にまとめています。 ## SQLModelのインストールとモデル定義：table=TrueとFieldの書き方 仮想環境を作って`pip install sqlmodel`を実行すると、SQLAlchemyとPydanticも一緒に入ります。uvを使うなら`uv add sqlmodel`です。非同期で使う場合は`pip install "sqlmodel[asyncio]"`とします。SQLAlchemy 2.1は非同期に必要なgreenletを既定では入れなくなったため、この`asyncio`エクストラで導入します。 `SQLModel`を継承して`table=True`を付けたクラスがテーブルになります。カラムの型は型ヒントから、制約は`Field()`の引数から決まります。 ``` from sqlalchemy import JSON from sqlmodel import Field, SQLModel, UniqueConstraint, create_engine class Hero(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str = Field(index=True, max_length=50) email: str = Field(unique=True) secret_name: str age: int | None = Field(default=None, index=True) settings: dict = Field(default_factory=dict, sa_type=JSON) class Membership(SQLModel, table=True): __tablename__ = "memberships" __table_args__ = ( UniqueConstraint("team_id", "user_id", name="uq_membership_team_user"), ) id: int | None = Field(default=None, primary_key=True) team_id: int user_id: int engine = create_engine("sqlite:///database.db") SQLModel.metadata.create_all(engine) ``` SQLiteで`create_all`が発行するDDLは次のとおりです（インデックスは別の`CREATE INDEX`で`ix_hero_name`と`ix_hero_age`が作られます）。 ``` CREATE TABLE hero ( id INTEGER NOT NULL, name VARCHAR(50) NOT NULL, email VARCHAR NOT NULL, secret_name VARCHAR NOT NULL, age INTEGER, settings JSON NOT NULL, PRIMARY KEY (id), UNIQUE (email) ) CREATE TABLE memberships ( id INTEGER NOT NULL, team_id INTEGER NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (id), CONSTRAINT uq_membership_team_user UNIQUE (team_id, user_id) ) ``` ### 主キーと自動採番：int | None と default=None の組み合わせ `id`を`int | None`にして`default=None`を与えるのは、Python側でオブジェクトを作った時点ではまだIDが無いからです。DDL上は`INTEGER NOT NULL`の主キーになり、INSERT時にDBが値を振ります。`id: int`とだけ書くと、FastAPIの入力スキーマなどに流用したときにIDが必須項目として扱われます。 型ヒントが`str | None`なら`NULL`可、`str`なら`NOT NULL`です。`max_length=50`は`VARCHAR(50)`に反映されますが、SQLiteはこの長さを強制しません。長さを指定しない`str`は、SQLiteでは長さなしの`VARCHAR`、MySQL方言では`VARCHAR(255)`になります。 ### Field(unique=True)とUniqueConstraint：単一列・複数列のユニーク制約 1列だけで一意にするなら`Field(unique=True)`で足り、DDLに`UNIQUE (email)`が入ります。「同じチームに同じユーザーを2回登録できない」のように複数列の組み合わせで一意にするときは、`Field`では書けないので、クラス変数`__table_args__`に`UniqueConstraint`を置きます。`UniqueConstraint`は`sqlmodel`から直接importできます。制約名（`name=`）を付けておくと、Alembicで制約を変更するときに名前で指定でき、エラーから原因の制約も特定しやすくなります。 カラムを`sa_column=Column(...)`で丸ごと指定した場合は、`Field`側の`unique`や`index`を併用できません。0.0.48でも`RuntimeError: Passing unique is not supported when also passing a sa_column`で止まるので、`Column(String(10), unique=True)`のようにColumn側へ書きます。 ### テーブル名とJSON列：\_\_tablename\_\_ と sa\_type の指定 テーブル名は既定でクラス名の小文字です（`Hero`は`hero`、`OrderItem`は`orderitem`）。スネークケースや複数形にしたいときは`__tablename__ = "memberships"`を書きます。外部キーの`foreign_key="memberships.id"`もこの名前で参照します。 `dict`や`list`は型ヒントだけではSQL型にならないので、`sa_type=JSON`でSQLAlchemyの型を指定します。この定義では`hero.settings["theme"] = "dark"`のような辞書内部の書き換えをORMが変更として検知せず、`commit()`しても保存されません（実行すると空の辞書のままでした）。更新時は新しい辞書を再代入するか、`sqlalchemy.ext.mutable`の`MutableDict`を使います。`sa_type`は型だけを差し替える引数なので、`sa_column`と違い`index`や`nullable`と併用できます。 ## エンジン接続とcreate\_allでのテーブル作成 ### 接続URL：SQLite・PostgreSQL・MySQLのドライバ指定 `create_engine`はSQLAlchemyの関数の再エクスポートで、URLの書式もSQLAlchemyに従います。PostgreSQLとMySQLはドライバを別に入れ、URLの`+`以降で指定します。 | DB | 接続URLの例 | 追加で入れるパッケージ | | ---------- | ---------------------------------------------- | ----------------- | | SQLite | `sqlite:///database.db` | 不要（標準ライブラリ） | | PostgreSQL | `postgresql+psycopg://user:pass@host:5432/app` | `psycopg[binary]` | | MySQL | `mysql+pymysql://user:pass@host:3306/app` | `pymysql` | `postgresql://`とだけ書いたときのドライバは、SQLAlchemyの版で変わります。2.1.6では`psycopg`（3系）、2.0系では`psycopg2`が選ばれます。Python 3.10では2.0系が入るので、ドライバは`+psycopg`のように明示してください。2系と3系の違いは[psycopg2とpsycopg3の違いと選び方](/tech/details/8225/)にまとめています。実行されるSQLを見たいときは`create_engine(url, echo=True)`とします。 ### create\_allの対象範囲：新規テーブル作成と既存スキーマ変更の違い `SQLModel.metadata.create_all(engine)`は、`SQLModel.metadata`に登録済みで、DBにまだ無いテーブルだけを作ります。モデルは`table=True`のクラスが**importされた時点で**登録されるので、モデルを別ファイルに分けているなら、`create_all`を呼ぶモジュールでimportしておかないとテーブルが作られません。 既存のテーブルには何もしません。モデルに列を足して`create_all`を再実行しても列は増えず、SQLiteならINSERT時の`OperationalError: table widget has no column named color`で初めて気づきます。データを残したままスキーマを変えるならAlembicを使います。SQLModelと組み合わせるときの注意は後述のFAQに、Alembic自体の使い方は[Alembicとは？マイグレーションの使い方とautogenerateの落とし穴](/tech/details/3417/)にあります。 ## SQLModelのCRUD：session.execとselectでの書き方 ### Create・Read：add/commitとwhere・in\_・or\_・getの使い分け 登録は`session.add()`して`session.commit()`、取得は`select()`を`session.exec()`に渡します。`session.exec`はSQLModel独自のメソッドで、`select(Hero)`の結果が`Hero`型として補完されます。 ``` from sqlmodel import Session, col, or_, select with Session(engine) as session: session.add(Hero(name="Deadpond", email="d@example.com", secret_name="Dive Wilson", age=30)) session.add(Hero(name="Rusty-Man", email="r@example.com", secret_name="Tommy Sharp", age=48)) session.add(Hero(name="Spider-Boy", email="s@example.com", secret_name="Pedro Parqueador")) session.commit() with Session(engine) as session: hero = session.exec(select(Hero).where(Hero.name == "Deadpond")).first() print(hero.id, hero.name) print([h.name for h in session.exec(select(Hero).where(col(Hero.age).in_([30, 48]))).all()]) print([h.name for h in session.exec( select(Hero).where(or_(col(Hero.age) > 40, col(Hero.age).is_(None))) ).all()]) print(session.get(Hero, 2).name, session.get(Hero, 99)) ``` ``` 1 Deadpond ['Deadpond', 'Rusty-Man'] ['Rusty-Man', 'Spider-Boy'] Rusty-Man None ``` `in_`や`is_`を呼ぶ列は`col()`で包みます。`Hero.age`は型ヒント上`int | None`なので、そのままでは型チェッカーが`in_`を知らずに警告を出すためで、実行時の動作は変わりません。`.where()`に条件を複数並べるとAND、ORは`or_()`です。`.first()`は0件なら`None`、`.one()`は0件でも2件以上でも例外、主キーで1件引くなら`session.get(Hero, 2)`が最短です。 SQLAlchemy 1.x流の`session.query(Hero)`も結果は返しますが、SQLModelの`Session`では「You probably want to use `session.exec()` instead of `session.query()`」という`DeprecationWarning`が発生します（Pythonの既定の警告フィルターでは、スクリプト本体から呼んだときに同じ箇所で1回だけ表示され、importしたモジュール内の呼び出しは表示されません）。古いサンプルは`session.exec(select(...))`に読み替えてください。 ### Update・Delete：1件ずつの変更と一括のupdate文・delete文 1件の更新は、取得したオブジェクトの属性を書き換えて`commit()`します。DB側の値を反映したオブジェクトが必要なら`session.refresh()`を呼びます。条件に合う行をまとめて変えるときは、`update`文・`delete`文を`session.exec`に渡します。どちらも`sqlmodel`からimportできます。 ``` from sqlmodel import delete, update with Session(engine) as session: hero = session.get(Hero, 1) hero.age = 31 session.add(hero) session.commit() session.refresh(hero) result = session.exec(update(Hero).where(col(Hero.age).is_(None)).values(age=0)) print("updated:", result.rowcount) result = session.exec(delete(Hero).where(Hero.name == "Spider-Boy")) print("deleted:", result.rowcount) session.commit() session.delete(session.get(Hero, 2)) session.commit() print(session.exec(select(Hero.name)).all()) ``` ``` updated: 1 deleted: 1 ['Deadpond'] ``` 一括`update`文の`synchronize_session`は既定が`"auto"`で、同じセッションに読み込み済みのオブジェクトにも値が反映されます。`hero`を保持したまま`update(Hero).where(Hero.id == 1).values(age=77)`を実行すると、`refresh()`なしで`hero.age`は77になりました。 ## datetime列の扱い：0.0.45からUTC前提に変わった書き込みと比較 0.0.45以降、型ヒントが`datetime`の列は`UTCDateTime`型（`DateTime(timezone=True)`が土台）になります。PostgreSQL方言のDDLは`TIMESTAMP WITH TIME ZONE`です。0.0.44までは`DateTime(timezone=False)`でした。 ``` from datetime import datetime, timedelta, timezone from pydantic import NaiveDatetime from sqlmodel import Field, Session, SQLModel, create_engine class Event(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) at: datetime class Legacy(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) at: NaiveDatetime engine = create_engine("sqlite:///events.db") SQLModel.metadata.create_all(engine) with Session(engine) as session: event = Event(at=datetime(2026, 10, 28, 9, 0, tzinfo=timezone(timedelta(hours=9)))) session.add(event) session.commit() session.refresh(event) print("read:", repr(event.at)) try: session.add(Event(at=datetime(2026, 10, 28, 9, 0))) session.commit() except Exception as e: session.rollback() print(type(e).__name__, str(e).splitlines()[0]) ``` ``` read: datetime.datetime(2026, 10, 28, 0, 0, tzinfo=datetime.timezone.utc) StatementError (builtins.ValueError) Datetime values must have timezone information. Use datetime.now(timezone.utc), or annotate the field with NaiveDatetime for naive storage. ``` 日本時間9時で保存した値は、UTCの0時として読み戻されます。タイムゾーン情報の無い`datetime(2026, 10, 28, 9, 0)`は、INSERTでも`where(Event.at > datetime(2026, 1, 1))`のような比較でも同じ`ValueError`になりました。`datetime.utcnow()`や`datetime.now()`で値を作っている既存コードは、0.0.45に上げた時点で書き込みが失敗します。 既存DBは自動では変わりません。naiveのまま保存を続けるなら、上の`Legacy`のように`NaiveDatetime`で注釈するか`Field(sa_type=DateTime(timezone=False))`を指定します。PostgreSQLで新しい型へ移るなら列のマイグレーション、SQLite・MySQLで移るならUTC以外で入っている値のデータ移行が要ります。手順は公式の[Datetimes and Timezones](https://sqlmodel.tiangolo.com/advanced/datetime/)にまとまっています。 ## ユニーク制約違反とバリデーション：table=Trueモデルの落とし穴 「`unique=True`なのに重複が入った」ときは、既存テーブルに制約が実際に付いているかを調べます（`create_all`は既存テーブルに制約を足しません）。「`field_validator`が呼ばれない」のは、`table=True`モデルを直接生成すると検証が走らないためです。 ### ユニーク制約違反：commit時のIntegrityErrorと応答処理 重複は`session.add()`の時点ではなく、SQLが発行される`commit()`（または`flush()`）で`sqlalchemy.exc.IntegrityError`として表面化します。例外が出たセッションは`rollback()`するまで次の操作ができません。 ``` from sqlalchemy.exc import IntegrityError with Session(engine) as session: session.add(Hero(name="Dup", email="d@example.com", secret_name="x")) try: session.commit() except IntegrityError as e: session.rollback() print(e.orig) ``` ``` UNIQUE constraint failed: hero.email ``` 複合ユニーク制約では`UNIQUE constraint failed: memberships.team_id, memberships.user_id`となり、どの組み合わせで衝突したかが分かります。事前に`select`で存在確認してからINSERTする方法は、2つのリクエストが同時に来ると両方が確認をすり抜けるので、制約違反の捕捉の代わりになりません。一意性の判定はDB制約に任せ、アプリ側は`IntegrityError`の原因を確認し、対象のユニーク制約違反だけを409などの応答に変換します。 ### table=Trueモデルの検証：直接生成とmodel\_validateの違い ``` hero = Hero(name=123, email=None, secret_name="x", age="abc") print(repr(hero.name), repr(hero.email), repr(hero.age)) try: Hero.model_validate({"name": "ok", "email": "e@x", "secret_name": "x", "age": "abc"}) except Exception as e: print(type(e).__name__) print(repr(Hero.model_validate({"name": "ok", "email": "e@x", "secret_name": "x", "age": "42"}).age)) ``` ``` 123 None 'abc' ValidationError 42 ``` 直接生成では型の合わない値がそのまま入り、`model_validate()`を通したときだけ検証と型変換が走ります。`field_validator`も同じで、前後の空白を削るバリデータを付けたテーブルモデルは、直接生成では呼ばれず`model_validate`では呼ばれました。外部から来るデータは、`table=True`を付けない入力用モデルで受けてから`Hero.model_validate()`で変換する経路に一本化します。 ## リレーションの定義：1対多・多対多とN+1の回避 この章の`Hero`は前章までと別の定義です。同じプロセスで続けて定義すると`Table 'hero' is already defined for this MetaData instance`で止まるので、別ファイル・新しいDBファイルで動かしてください。 ### 1対多：foreign\_keyとRelationship(back\_populates)の対応 子（多側）に`foreign_key="team.id"`の列を置き、両側に`Relationship`を定義して`back_populates`で互いの属性名を指し合います。`ondelete`を付けると外部キーに`ON DELETE`句が入ります。SQLiteで外部キーの検査や削除時の動作を有効にするには、接続ごとに`PRAGMA foreign_keys=ON`が要ります。 ``` from sqlmodel import Field, Relationship, SQLModel class HeroMissionLink(SQLModel, table=True): hero_id: int | None = Field(default=None, foreign_key="hero.id", primary_key=True) mission_id: int | None = Field(default=None, foreign_key="mission.id", primary_key=True) class Team(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str = Field(unique=True) heroes: list["Hero"] = Relationship(back_populates="team") class Hero(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str team_id: int | None = Field(default=None, foreign_key="team.id", ondelete="SET NULL") team: Team | None = Relationship(back_populates="heroes") missions: list["Mission"] = Relationship(back_populates="heroes", link_model=HeroMissionLink) class Mission(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) title: str heroes: list[Hero] = Relationship(back_populates="missions", link_model=HeroMissionLink) ``` ``` CREATE TABLE hero ( id INTEGER NOT NULL, name VARCHAR NOT NULL, team_id INTEGER, PRIMARY KEY (id), FOREIGN KEY(team_id) REFERENCES team (id) ON DELETE SET NULL ) ``` まだ定義していないクラスを参照する型ヒントは`list["Hero"]`のように文字列にします。`Relationship`はDBの列を作らず、`team.heroes`や`hero.team`でオブジェクトを辿るためのORM上の属性です。 ### 多対多リレーション：link\_modelによる中間テーブルの指定 多対多は、両方の外部キーを複合主キーにした中間テーブル（上の`HeroMissionLink`）を定義し、両側の`Relationship`に`link_model`を渡します。関連付けは`hero.missions.append(mission)`、解除は`hero.missions.remove(mission)`で、`commit()`すると中間テーブルの行がINSERT・DELETEされます。中間テーブルに「参加日」のような独自の列を持たせるなら、`link_model`をやめ、中間テーブル自体を1対多2本で扱う設計に切り替えます。 ### selectinloadとjoin：関連データ取得時のSQL本数と用途の違い リレーション属性は既定で遅延読み込みです。10チーム・20人を登録して全チームの`heroes`を参照すると、発行されたSQLは次の本数でした。 ``` from sqlalchemy.orm import selectinload teams = session.exec(select(Team)).all() sum(len(t.heroes) for t in teams) # 発行SQL: 11本（チーム一覧1本＋チームごと10本） teams = session.exec(select(Team).options(selectinload(Team.heroes))).all() sum(len(t.heroes) for t in teams) # 発行SQL: 2本（チーム一覧1本＋IN句でヒーロー一括1本） ``` 一覧APIでチームごとの所属ヒーローを返す処理は、件数に比例してSQLが増えるN+1になります。`selectinload`を付ければこの例では2本です。親の主キーは最大500件ずつに分けて取得されるため、チームを501件にすると3本になりました（[SQLAlchemy 2.1のリレーション読み込みの解説](https://docs.sqlalchemy.org/en/21/orm/queryguide/relationships.html)）。関連オブジェクトが要らず条件で絞るだけなら、`select(Hero, Team).join(Team).where(Team.name == "team1")`で`(Hero, Team)`のタプルとして受け取ります。 なお、`with Session(engine)`を抜けた後に`team.heroes`を初めて参照すると、遅延読み込みのSQLを発行するセッションが無いため`DetachedInstanceError`になります。ブロックの外で使う関連データは`selectinload`で先に読み込んでおきます。 ## FastAPIとの連携：lifespan・SessionDep・入出力モデルの分離 SQLModelの公式チュートリアル（[Multiple Models with FastAPI](https://sqlmodel.tiangolo.com/tutorial/fastapi/multiple-models/)）も、テーブルモデルをそのままリクエストボディに使う形から、`HeroBase`を継承した`HeroCreate`・`HeroPublic`に分ける形へ進みます。DBが採番する`id`をクライアントに指定させないためです。下のコードはさらに、`email`と`secret_name`を`HeroPublic`に含めずレスポンスから外しています。前章までとは別の`main.py`として、新しいDBファイル`app.db`で動かす例です（`settings`列を持つ最初の`database.db`を使い回すと、NOT NULL違反で登録に失敗します）。 ``` from contextlib import asynccontextmanager from typing import Annotated from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.exc import IntegrityError from sqlmodel import Field, Session, SQLModel, create_engine class HeroBase(SQLModel): name: str = Field(index=True) age: int | None = Field(default=None, index=True) class Hero(HeroBase, table=True): id: int | None = Field(default=None, primary_key=True) email: str = Field(unique=True) secret_name: str class HeroCreate(HeroBase): email: str secret_name: str class HeroPublic(HeroBase): id: int class HeroUpdate(SQLModel): name: str | None = None age: int | None = None engine = create_engine("sqlite:///app.db", connect_args={"check_same_thread": False}) def get_session(): with Session(engine) as session: yield session SessionDep = Annotated[Session, Depends(get_session)] @asynccontextmanager async def lifespan(app: FastAPI): SQLModel.metadata.create_all(engine) yield app = FastAPI(lifespan=lifespan) @app.post("/heroes/", response_model=HeroPublic) def create_hero(hero: HeroCreate, session: SessionDep): db_hero = Hero.model_validate(hero) session.add(db_hero) try: session.commit() except IntegrityError as e: session.rollback() if "UNIQUE constraint failed: hero.email" not in str(e.orig): raise raise HTTPException(status_code=409, detail="email already exists") session.refresh(db_hero) return db_hero @app.patch("/heroes/{hero_id}", response_model=HeroPublic) def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep): db_hero = session.get(Hero, hero_id) if not db_hero: raise HTTPException(status_code=404, detail="Hero not found") if "name" in hero.model_fields_set and hero.name is None: raise HTTPException(status_code=422, detail="name must not be null") db_hero.sqlmodel_update(hero.model_dump(exclude_unset=True)) session.add(db_hero) session.commit() session.refresh(db_hero) return db_hero ``` TestClientで順にリクエストを送った結果です。奇数行が送ったリクエストの要約、偶数行がステータスコードとレスポンス本文です（422の1件目は`detail`のエラー種別だけを抜き出しています）。 ``` POST /heroes/ {"name": "Deadpond", "email": "d@example.com", "secret_name": "Dive Wilson"} 200 {'name': 'Deadpond', 'age': None, 'id': 1} POST /heroes/ 同じemailでもう一度 409 {'detail': 'email already exists'} POST /heroes/ {"age": "abc", ...} 422 int_parsing PATCH /heroes/1 {"age": 30} 200 {'name': 'Deadpond', 'age': 30, 'id': 1} PATCH /heroes/1 {"name": null} 422 {'detail': 'name must not be null'} PATCH /heroes/9 {"age": 30} 404 {'detail': 'Hero not found'} ``` ### 起動時のテーブル作成：on\_eventの非推奨化とlifespanへの移行 起動時の`create_all`を`@app.on_event("startup")`で書く例は、SQLModel公式の最初のFastAPIページやFastAPI公式の日本語版チュートリアルにも残っていますが、FastAPI 0.162.0では登録した時点で「on\_event is deprecated, use lifespan event handlers instead.」という`DeprecationWarning`が出ます。`asynccontextmanager`で包んだ関数の`yield`より前が起動時、後が終了時の処理で、`FastAPI(lifespan=lifespan)`で登録します（[FastAPIのLifespan Events](https://fastapi.tiangolo.com/advanced/events/)）。本番でAlembicを使うなら、ここで`create_all`は呼ばずマイグレーションに任せます。 ### セッションの依存性注入と入出力モデル：SessionDep・HeroCreate・HeroUpdate `get_session`は`with`の中で`yield`するので、FastAPIがリクエストごとにセッションを開き、レスポンス後に閉じます。`check_same_thread=False`はSQLite専用の引数で、`def`のエンドポイントがスレッドプールで動くために指定します。1つのセッションを複数のリクエストで共有しないことは、`get_session`の構造で守ります。 `HeroCreate`は`id`を持たないので、クライアントがIDを送っても無視されます。部分更新の`HeroUpdate`は全項目を省略可能にし、`model_dump(exclude_unset=True)`で送られた項目だけを取り出して`sqlmodel_update()`で反映します。`exclude_unset`を忘れると送っていない項目が`None`で上書きされます。また、公式チュートリアルと同じ`name: str | None = None`のままだと`{"name": null}`が通り、NOT NULL違反の500になるため、上のコードでは`model_fields_set`で422に変えています。ディレクトリ分割やテストを含む構成は[FastAPIのベストプラクティス（ディレクトリ構成・設計・テスト）](/tech/details/4347/)が参考になります。 ### 非同期の落とし穴：async defと同期Session・AsyncSessionの違い エンドポイントを`async def`にして中で同期の`Session`を使うと、DBの応答を待つ間イベントループが止まり、ほかのリクエストも待たされます。SQLiteに登録したsleep関数で0.5秒待つクエリを作り、httpxの`ASGITransport`経由で10件を同時に送って計測すると、次の差が出ました。ネットワークを通らない単一プロセス内の比較で、数値は環境によって変わります（再実行では5.84秒・0.78秒・0.89秒）が、`async def`＋同期`Session`だけが件数分直列になる傾向は同じです。 | エンドポイントの書き方 | 10件同時の所要時間 | | ------------------------------------- | ---------- | | `async def`＋同期`Session` | 6.07秒 | | `def`＋同期`Session` | 0.73秒 | | `async def`＋`AsyncSession`（aiosqlite） | 0.68秒 | 同期の`Session`を使うなら`def`で書けば、FastAPIがスレッドプールで並行に動かします。`async def`で書くなら、`pip install "sqlmodel[asyncio]"`と非同期ドライバ（aiosqlite、asyncpgなど）を入れ、`sqlmodel.ext.asyncio.session`の`AsyncSession`と`create_async_engine`を組み合わせます。 ``` from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession async_engine = create_async_engine("sqlite+aiosqlite:///app.db") async def list_heroes(): async with AsyncSession(async_engine) as session: result = await session.exec(select(Hero)) return result.all() ``` `AsyncSession`では、`hero.team`のような通常の属性参照で遅延読み込みのSQLを暗黙に発行できません。SQLAlchemyの`AsyncAttrs`を混ぜて`awaitable_attrs`でawaitする方法もありますが、関連データは`selectinload`で先に読み込んでおくのが簡単です（[SQLAlchemyのasyncio拡張の解説](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession)）。FastAPIの`async def`と`def`の使い分けは[FastAPIとは｜最短で動かす手順とasync/defの使い分け](/tech/details/3859/)でも扱っています。 DB設計やスキーマ移行まで含めて相談したい場合は、[データベース設計・移行支援](/service/system/database/)で対応しています。 ## よくある質問 ### Field(unique=True, index=True) と両方付けると何が変わりますか？ テーブル定義の`UNIQUE`制約ではなく、ユニークインデックスとして作られます。`email: str = Field(unique=True, index=True)`のテーブル`u`を0.0.48＋SQLiteで作ると、`CREATE TABLE`から`UNIQUE (email)`が消え、`ix_u_email`というユニークインデックスが作られました。重複を拒否する効果は同じですが、Alembicの差分は「制約」ではなく「インデックス」として出ます。 ### SQLModelでAlembicのautogenerateを使うと失敗するのはなぜですか？ 生成されたファイルに`sqlmodel.sql.sqltypes.AutoString()`が書かれるのに、`import sqlmodel`が入らないためです。Alembic 1.20.1でも`alembic upgrade head`が`NameError: name 'sqlmodel' is not defined`で止まりました。0.0.45以降は`datetime`列も`sqlmodel.sql.sqltypes.UTCDateTime()`と書かれます。`migrations/script.py.mako`に`import sqlmodel`を足し、`env.py`の`target_metadata`に`SQLModel.metadata`を渡せば、以降の生成ファイルは通ります。 ### session.query()はSQLModelでも使えますか？ 動きますが、SQLModelの`Session`では`DeprecationWarning`が出ます。型の補完が効く`session.exec(select(Model))`が推奨で、公式チュートリアルもこの書き方で統一されています。生のSQLAlchemyの結果行が欲しいときは`session.execute()`を使います。 ### 複合主キーはどう書きますか？ 主キーにしたい複数の列それぞれに`Field(primary_key=True)`を付けます。`order_id`と`line_no`の両方に付けたモデルでは、DDLが`PRIMARY KEY (order_id, line_no)`になりました。別にサロゲートキーを持ったうえで組み合わせを一意にしたいだけなら、`__table_args__`の`UniqueConstraint`を使います。 ### PostgreSQLのJSONB列はどう定義しますか？ `from sqlalchemy.dialects.postgresql import JSONB`をimportし、`data: dict = Field(default_factory=dict, sa_type=JSONB)`と書きます。PostgreSQL方言のDDLは`data JSONB NOT NULL`です。テストだけSQLiteで回す構成では同じモデルが使えないので、DBを切り替える前提なら汎用の`sqlalchemy.JSON`を選びます。 ## 関連記事 - [SQLAlchemyとは？読み方・使い方とPython ORMの基本を2.1.5の実行結果で解説](/tech/details/7912/) - [ORMとは？O/Rマッパーの仕組み・メリットとデメリットをSQL比較と実行結果で解説](/tech/details/3755/) - [Pydanticとは？読み方・BaseModelの使い方とV1との違い【2.13対応】](/tech/details/4838/) - [Alembic（アレンビック）とは？SQLAlchemyのマイグレーションの使い方とautogenerateの落とし穴【1.20対応】](/tech/details/3417/) - [FastAPIのベストプラクティス｜ディレクトリ構成・SQLAlchemy 2.1の設計・テストを実測で解説](/tech/details/4347/) --- 出典: [SQLModelとは？使い方・モデル定義・unique制約・FastAPI連携【0.0.48対応】]()（株式会社一創）