23 人が閲覧(直近 30 日) Python

SQLModelとは?使い方・モデル定義・unique制約・FastAPI連携【0.0.48対応】

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公式ドキュメント)。DBとのやり取りはSQLAlchemyのエンジンとセッションがそのまま担い、SQLModelはそこにPydanticの型検証と、1クラスで両方を表す仕組みを足しています。ORMそのものの仕組みはORMとは?O/Rマッパーの仕組み・メリットとデメリットで解説しています。

二重定義の解消:テーブルとAPI入出力モデルのフィールド共有

SQLAlchemyだけでAPIを作ると、テーブル用のDeclarativeBaseのクラスと、リクエストやレスポンスを検証するPydanticのBaseModelを別々に書き、同じ「名前・年齢」を2か所で宣言して相互に変換するコードが要ります。SQLModelでは、SQLModelを継承したクラスの型ヒントがそのままカラム型にもPydanticの検証ルールにもなります。table=Trueを付けたクラスがテーブル、付けないクラスは検証専用のモデルです。共通の項目を基底クラスに置いて両者で継承すれば、フィールドの宣言は1回で済みます(後述のFastAPI章のHeroBaseがこの形です)。

FastAPIは必須ではありません。SQLModel単体でもスクリプトやバッチのORMとして使えます。ただし、入出力モデルとテーブルを継承で共有できる利点はAPIを作るときに最も効くので、FastAPIとの連携は代表的な利用例です。Pydantic側の書き方はPydanticとは?BaseModelの使い方とV1との違いが参考になります。

開発状況:0.0.48・Python 3.10以上・依存するSQLAlchemyの範囲

PyPIの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の変更はリリースノートで「Breaking Changes」と明記されています。1.0に達していないライブラリは、マイナー相当の変更が0.0.xの末尾で入ります。本番ではsqlmodel==0.0.48のように版を固定し、上げる前に公式のリリースノートを読む運用が前提です。

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の基本にまとめています。

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の違いと選び方にまとめています。実行される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の落とし穴にあります。

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="[email protected]", secret_name="Dive Wilson", age=30))
    session.add(Hero(name="Rusty-Man", email="[email protected]", secret_name="Tommy Sharp", age=48))
    session.add(Hero(name="Spider-Boy", email="[email protected]", 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にまとまっています。

ユニーク制約違反とバリデーション: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="[email protected]", 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のリレーション読み込みの解説)。関連オブジェクトが要らず条件で絞るだけなら、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)も、テーブルモデルをそのままリクエストボディに使う形から、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": "[email protected]", "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)。本番で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のベストプラクティス(ディレクトリ構成・設計・テスト)が参考になります。

非同期の落とし穴: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拡張の解説)。FastAPIのasync defとdefの使い分けはFastAPIとは|最短で動かす手順とasync/defの使い分けでも扱っています。

DB設計やスキーマ移行まで含めて相談したい場合は、データベース設計・移行支援で対応しています。

よくある質問

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を選びます。

関連記事

お気に入りに入れた記事の一覧

一創の公式 SNS

X では毎朝の新着記事と IT ニュースの速報を、YouTube・TikTok・Instagram・LinkedIn・Facebook では記事の解説動画を配信しています。

この記事は以下の記事からリンクされています

資料請求

今日のトレンド記事 直近 24 時間で、いつもより多く読まれている記事

  1. 2026.11.09 テックブログ エディオン不正アクセス第2報:カード番号の対象確認とログ・決済画面の改ざん対策
  2. 2026.11.09 テックブログ LANSCOPE不正アクセス第2報:製品への影響と偽連絡の確認、停止中のライセンス認証の窓口
  3. 2026.11.10 テックブログ Claude Fable 5.2とは:料金・破壊的変更4点とFable 5.1からの移行手順【2026年11月】
  4. 2026.11.10 テックブログ ニトリの情報漏洩:最大3,762万件の対象確認となりすまし連絡の見分け方
  5. 2026.11.09 テックブログ ステップゴルフ不正アクセス:口座情報の対象確認と偽連絡・引き落としの注意点

RELATED POSTS 関連記事

目次