Python

SQLModelのモデル定義とテーブル作成|unique制約・CRUD・リレーション・FastAPI連携【0.0.42】

SQLModelのモデル定義とテーブル作成|unique制約・CRUD・リレーション・FastAPI連携【0.0.42】

SQLModelでテーブルを作るときに迷うのは、Field()のどの引数がどんなDDLになるか、ユニーク制約をどこに書くか、そしてFastAPIと組み合わせたときにどこまで同じクラスを使い回してよいかです。本記事では、SQLModel 0.0.42(2026年8月28日リリース)とSQLAlchemy 2.0.52・Pydantic 2.13.5・FastAPI 0.141.1の組み合わせで各コードを実行し、生成されたDDLやエラーメッセージをそのまま載せています。

SQLModelが何者か、SQLAlchemyとどちらを選ぶかはSQLModelの特徴とSQLAlchemyとの使い分けで扱っているので、ここでは実装の手順と落とし穴に絞ります。

まとめ:SQLModelでテーブルを作り操作するときの要点

  • 主キーはid: int | None = Field(default=None, primary_key=True)。INSERT時にDB側で採番されます。
  • 単一列のユニーク制約はField(unique=True)、複数列の組み合わせは__table_args__にUniqueConstraintを置きます。
  • SQLModel.metadata.create_all(engine)は無いテーブルを作るだけで、既存テーブルへの列追加はしません。変更はAlembicで管理します。
  • クエリはsession.exec(select(...))で書き、一括更新・削除はupdate・delete文をsession.execに渡します。
  • table=Trueのモデルはインスタンス生成時にバリデーションが走りません。外部入力はtable=Trueでないモデルで受けてmodel_validateで変換します。
  • FastAPIではon_eventではなくlifespanでテーブルを作り、セッションは依存性注入でリクエストごとに開きます。

モデル定義からバリデーションまでのコードは、空の作業用DBで順に実行します。リレーション章とFastAPI章はそれぞれ独立したサンプルなので、別プロセス・別の作業ディレクトリで実行してください。同じプロセスではHeroの定義が重複し、同じdatabase.dbを使うと既存テーブルと列構成が衝突します。

SQLModelのモデル定義:table=TrueとFieldで決まるカラムと制約

SQLModelを継承しtable=Trueを付けたクラスがテーブルになります。付けなければ単なるPydanticモデルで、DBには対応しません。カラムの型は型ヒントから、制約は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とだけ書くと、Hero(name=...)の段階でIDを渡す前提の型になり、FastAPIの入力スキーマなどに流用したときに必須項目として扱われます。

型ヒントがstr | NoneならNULL可、strならNOT NULLになります。上のageだけがNOT NULLを持たないのはこのためです。max_length=50はVARCHAR(50)に反映されますが、SQLiteはこの長さ指定を強制しません。50文字以内に制限するには入力モデルの検証などが必要です。また、指定しないstrはSQLiteでは長さなしのVARCHARになります。SQLModelの既定の文字列型は、MySQLではVARCHAR(255)になります。

Field(unique=True)とUniqueConstraint:単一列・複数列の指定

1列だけで一意にしたいならField(unique=True)で足ります。DDLにはUNIQUE (email)が入ります。「同じチームに同じユーザーを2回登録できない」のように複数列の組み合わせで一意にしたい場合は、Fieldでは書けないので、クラス変数__table_args__にSQLAlchemyのUniqueConstraintを置きます。UniqueConstraintはsqlmodelから直接importできます。

制約名(name=)は省略できますが、付けておくとAlembicで制約を削除・変更するときに名前で指定でき、エラーメッセージから原因の制約も特定しやすくなります。

注意点が1つあります。カラムをsa_column=Column(...)で丸ごと指定した場合、Field側のuniqueやindexは併用できません。0.0.42では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)。スネークケースや複数形にしたいときは、Membershipの例のように__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 2.0に従います。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://とだけ書くとドライバはpsycopg2が選ばれます。新規にpsycopg 3を使うなら+psycopgを明示してください。2系と3系の違いはpsycopg2とpsycopg3の非互換と選び方にまとめています。開発中に実行SQLを確認したいときはcreate_engine(url, echo=True)とします。

create_allの対象範囲:新規テーブル作成と既存スキーマ変更の違い

SQLModel.metadata.create_all(engine)は、SQLModel.metadataに登録済みで、かつDBにまだ無いテーブルだけをCREATE TABLEします。モデルはtable=Trueのクラスがimportされた時点でmetadataに登録されるので、モデルを別ファイルに分けている場合はcreate_allを呼ぶモジュールでimportしておかないとテーブルが作られません。

既に存在するテーブルには何もしません。モデルに列を足してcreate_allを再実行しても、既存テーブルに列は増えず、SQLiteならINSERT時にOperationalError: table widget has no column named colorのようなエラーで初めて気づくことになります。データを残したままスキーマを変えるならAlembicでマイグレーションを管理します。SQLModelと組み合わせるときの落とし穴は後述のFAQに、Alembic自体の使い方はAlembicによるマイグレーション管理の基本にあります。

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型として補完されます。SQLAlchemy 1.x流のsession.query()は使わずに済みます。

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)が最短で、存在しなければNoneです。

Update・Delete:1件ずつの変更と一括のupdate文・delete文

1件の更新は、取得したオブジェクトの属性を書き換えてcommit()します。DB側の値(採番やデフォルト値)を反映したオブジェクトが必要ならsession.refresh()を呼びます。条件に合う行をまとめて変えるときは、オブジェクトを1件ずつ読み込まずにupdate文・delete文を発行します。どちらもsqlmodelからimportでき、session.execに渡せます。

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']

SQLAlchemy 2.0の一括update文はsynchronize_sessionの既定値が"auto"で、同じセッションに読み込み済みのオブジェクトにも値を反映します。ageが31のheroを保持したままupdate(Hero).where(Hero.id == 1).values(age=77)を実行すると、refresh()なしでhero.ageは77になりました。件数が多い一括更新で1件ずつsession.getしてループする必要はありません。

ユニーク制約違反とバリデーション:table=Trueモデルの落とし穴

「unique=Trueを付けたのに重複データが入ってしまった/エラーの形が想定と違う」「field_validatorを書いたのに呼ばれない」という2つの相談は、原因が別々です。重複の拒否はDBの制約が行うので、重複が保存できた場合は、既存テーブルにユニーク制約が実際に付いているかを調べます(前述のとおりcreate_allは既存テーブルに制約を追加しません)。field_validatorが呼ばれないのは、table=Trueモデルを直接生成したときに検証が走らないためで、model_validateを通すかどうかの違いです。

ユニーク制約違反:flush・commit時のIntegrityErrorと応答処理

unique=Trueの検査はDBが行うので、重複は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の違い

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

field_validatorも同じ扱いです。nameの前後の空白を削るバリデータを付けたテーブルモデルで試すと、Doc(name=" a ")では呼ばれず値は' a 'のまま、Doc.model_validate({"name": " a "})では呼ばれて'a'になりました。外部から来るデータは、後述のFastAPI連携章のようにtable=Trueを付けない入力用モデルで受けてからHero.model_validate()でテーブルモデルに変換する、という経路に一本化すると、検証漏れが起きません。

リレーションの定義:1対多・多対多とN+1の回避

1対多:foreign_keyとRelationship(back_populates)の対応

子(多側)にforeign_key="team.id"の列を置き、両側にRelationshipを定義してback_populatesで互いの属性名を指し合います。ondeleteを付けると、DDLの外部キーにON DELETE句が入ります。SQLiteでDB側の外部キー検査や削除時の動作を有効にするには、各接続で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=HeroMissionLinkを渡します。関連付けは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(sqlalchemy.ormからimport)を付ければ、この10チームの例では2本で済みます。SQLAlchemy 2.0では親の主キーを最大500件ずつに分けて関連データを取得するため、親が501件なら合計3本になります。条件で絞り込むだけで関連オブジェクトの読み込みが不要なら、select(Hero, Team).join(Team).where(Team.name == "team1")のようにjoinして、(Hero, Team)のタプルとして受け取ります。

FastAPIとの連携:lifespan・SessionDep・入出力モデルの分離

SQLModelの公式チュートリアル(Multiple Models with FastAPI)も、テーブルモデルをそのままリクエストボディに使う形から、HeroBaseを継承したHeroCreate・HeroPublicに分ける形へ進みます。DBが採番するidをクライアントに指定させないためです。下のコードはこれに加えて、emailとsecret_nameをHeroPublicに含めず、レスポンスに出ないようにしています。

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:///database.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:
        session.rollback()
        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は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/9 {"age": 30}
404 {'detail': 'Hero not found'}

起動時のテーブル作成:on_eventは非推奨でlifespanへ

起動時にcreate_allを呼ぶ処理を@app.on_event("startup")で書く例は、SQLModel公式チュートリアルの最初のFastAPIページ(Simple Hero API)を含め今も多く残っていますが、FastAPI 0.141.1では登録した時点で「on_event is deprecated, use lifespan event handlers instead.」というDeprecationWarningが出ます。asynccontextmanagerで包んだ関数のyieldより前が起動時、後が終了時の処理になり、FastAPI(lifespan=lifespan)で登録します。本番でAlembicを使う構成では、ここでcreate_allを呼ばずマイグレーションに任せます。

セッションの依存性注入:SessionDepによるリクエスト単位の開閉

get_sessionはwithの中でyieldするジェネレータなので、FastAPIがリクエストごとにセッションを開き、レスポンス後に閉じます。Annotated[Session, Depends(get_session)]をSessionDepという名前にしておくと、各エンドポイントはsession: SessionDepと書くだけで済みます。SQLiteでcheck_same_thread=Falseを指定しているのは、FastAPIでは1つのリクエストが複数のスレッドにまたがって処理されうるためで、SQLite以外のドライバには無い引数です。代わりに「1つのセッションを複数のリクエストで共有しない」ことを、get_sessionの構造で守ります。ディレクトリ分割やテストまで含めた構成は、SQLAlchemy版ですがFastAPIのベストプラクティス(ディレクトリ構成・モデル設計・テスト)が参考になります。

入力・出力・部分更新のモデル:HeroCreate・HeroPublic・HeroUpdate

HeroCreateはidを持たないので、クライアントがIDを送っても無視されます。response_model=HeroPublicにより、テーブルにあるemailやsecret_nameはレスポンスから落ちます。部分更新のHeroUpdateは全項目を省略可能にし、model_dump(exclude_unset=True)で「送られてきた項目だけ」を取り出してsqlmodel_update()で反映します。exclude_unsetを忘れると、送っていない項目がNoneで上書きされます。

テーブルモデルを直接リクエストボディに使うと、前章のとおりバリデーション経路が曖昧になるうえ、idを外部から指定できてしまいます。学習用の最小例を除き、テーブルモデルのHeroとは別に、入力用HeroCreate・出力用HeroPublic・部分更新用HeroUpdateを定義してください。

よくある質問

Field(unique=True, index=True) と両方付けると何が変わりますか?

テーブル定義のUNIQUE制約ではなく、ユニークインデックスとして作られます。email: str = Field(unique=True, index=True)を持つテーブルuをSQLModel 0.0.42+SQLiteで作ると、CREATE TABLEからUNIQUE (email)が消え、代わりにix_u_emailというunique=Trueのインデックスが作られました。重複を拒否する効果は同じですが、Alembicでの差分は「制約」ではなく「インデックス」として出るので、あとから片方を外すとマイグレーションの内容が変わります。

SQLModelでAlembicのautogenerateを使うと失敗するのはなぜですか?

生成されたマイグレーションファイルにsqlmodel.sql.sqltypes.AutoString()が書かれるのに、import sqlmodelが入らないためです。そのままalembic upgrade headを実行するとNameError: name 'sqlmodel' is not definedで止まります。migrations/script.py.makoのimport sqlalchemy as saの下にimport sqlmodelを足し、env.pyのtarget_metadataにSQLModel.metadataを渡します。これで以降の生成ファイルのimport不足は解消できますが、autogenerateの出力は候補なので、適用前に変更内容を確認してください。

複合主キーはどう書きますか?

主キーにしたい複数の列それぞれにField(primary_key=True)を付けます。order_id: int = Field(primary_key=True)とline_no: int = Field(primary_key=True)を持つモデルでは、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になります。JSONBはPostgreSQL専用の型なので、テストだけSQLiteで回す構成では同じモデルが使えません。DBを切り替える前提なら汎用のsqlalchemy.JSONを選ぶ方が扱いやすくなります。

セッションを閉じた後にリレーション属性を参照するとエラーになるのはなぜですか?

遅延読み込みのSQLを発行するためのセッションが既に無いからです。with Session(engine)のブロックを抜けた後にteam.heroesを参照すると、DetachedInstanceError(Parent instance is not bound to a Session)になります。セッション内で一度参照しておくか、select(Team).options(selectinload(Team.heroes))で取得時に読み込んでおけば、ブロックの外でも値を使えます。

関連記事

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

資料請求

RELATED POSTS 関連記事

目次