SQLModelを調べて使ってみる

ikepon

FastAPI で使ってるので調べて、ローカルで使ってみる

https://sqlmodel.tiangolo.com/

ikepon

SQLModelはPythonコードからPythonオブジェクトを使ってSQLデータベースを操作するためのライブラリです。
- Python の型アノテーションをベースとしており、Pydantic と SQLAlchemy を利用しています。

ikepon

とりあえす動かしてみる

.
├── README.md
├── database.db // ここにサンプルデータを作成する
├── main.py // 今は使わない
├── pyproject.toml // uv add で sqlmodel を入れておく
├── scripts/
│   └── ipython_startup.py // ipython で動かす
├── src/
│   └── sqlmodel/
│       ├── database.py
│       ├── init_db.py
│       └── models.py
└── uv.lock

# models.py: サンプルにあるモデルを定義
from typing import Optional
from sqlmodel import SQLModel, Field

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

# database.py: sqlite で
from sqlmodel import SQLModel, create_engine

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

# init_db.py
from sqlmodel import Session
from .database import engine, create_db_and_tables
from .models import Hero

def init_db():
    create_db_and_tables()

    with Session(engine) as session:
        heroes = [
            Hero(name="Deadpond", secret_name="Dive Wilson"),
            Hero(name="Spider-Boy", secret_name="Pedro Parqueador"),
            Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48),
        ]

        for hero in heroes:
            session.add(hero)

        session.commit()

if __name__ == "__main__":
    init_db()

# ipython_startup.py: ipython で操作するための記述

from sqlmodel import Session, select
from src.sqlmodel.database import engine
from src.sqlmodel.models import Hero

print("✅ Load ipython_startup.py")


# SQLModelのサンプルコード
print("\n=== SQLModel Sample Code ===")
print("以下の変数が利用可能です：")
print("- session: SQLModelのセッション")
print("- Hero: ヒーローモデル")
print("- select: SQLModelのselect関数")

# セッションの作成
session = Session(engine)

# サンプルクエリの実行
print("\nヒーロー一覧の取得:")
heroes = session.exec(select(Hero)).all()
for hero in heroes:
    print(f"ID: {hero.id}, Name: {hero.name}, Secret Name: {hero.secret_name}, Age: {hero.age}")

print("\n=== サンプルクエリ ===")
print("""
# 全ヒーローの取得
heroes = session.exec(select(Hero)).all()

# 特定のヒーローの取得
hero = session.exec(select(Hero).where(Hero.name == "Deadpond")).first()

# 新しいヒーローの追加
new_hero = Hero(name="New Hero", secret_name="Secret Identity")
session.add(new_hero)
session.commit()

# ヒーローの更新
hero = session.exec(select(Hero).where(Hero.name == "Deadpond")).first()
hero.age = 30
session.add(hero)
session.commit()

# ヒーローの削除
hero = session.exec(select(Hero).where(Hero.name == "New Hero")).first()
session.delete(hero)
session.commit()
""")

初期データ投入

python -m src.sqlmodel.init_db

あとは仮想環境を立ち上げて

uv run ipython -i scripts/ipython_startup.py --no-confirm-exit

で操作できる

ikepon

データ作成

Model を作って、それを session.add(Model) -> session.commit() すれば保存される
engine
- DB と client の間にパイプを通して繋げるイメージ
session
- engine(パイプ)を通してDBを操作する窓口みたいなもの
- sessionを使ってデータのCRUDを実行する

hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)


engine = create_engine("sqlite:///database.db")


# この時点で Hero を特に指定してないけど、 hero テーブルが作成される
# これは事前に `class Hero(SQLModel, table=True):` を定義しているから SQLModel.metadata に登録されるから
# なので、ここで Hero とかを引数に特に取らなくて大丈夫な仕組みになってる
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.commit()

ikepon

You can even create SQLModel models that do not represent SQL tables. In that case, they would be the same as Pydantic models.
This is useful, in particular, because now you can create a SQL database model that inherits from another non-SQL model. You can use that to reduce code duplication a lot. It will also make your code more consistent, improve editor support, etc.

SQLModel はもともと：
• Pydantic（API用のバリデーションモデル）
• SQLAlchemy（DBテーブルのORMモデル）

この2つを組み合わせて、「API用にもDB用にも使えるモデル」を作れるライブラリ
でも、必ずしも全部のモデルを「テーブル」にする必要はない
FastAPI のモデル定義でよくみる XxxBase みたいなのを作って継承できるってのが言いたいこと

from sqlmodel import SQLModel, Field

# これは table=True がないからDBのテーブルではない
class HeroBase(SQLModel):
    name: str
    secret_name: str
    age: int | None = None

# テーブルでない HeroBase を継承して、最終的なテーブルを定義してる
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

これによって、POST だと id なしとかを正確に定義できるようになる

ikepon

 SQLModel完全入門：基礎から複雑なリレーションまで
 1. SQLModelの基礎SQLModel は、Python の型ヒントを活かして SQL データベースと連携するためのライブラリです。

FastAPIの作者である Sebastián Ramírez 氏によって作られ、以下の技術を統合しています：

SQLAlchemy：強力なORM機能

Pydantic：型安全・データバリデーション

 特徴型安全・自動補完に優れる
同じモデルでバリデーションとORMの両方を実現
FastAPIとの親和性が非常に高い

ikepon

2. SQLModel による定義

2.1 1テーブルの定義

from typing import Optional
from sqlmodel import SQLModel, Field

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

table=True により、このモデルがデータベーステーブルになることを表します。

2.2 1 対多（One to Many）

「1対多（One to Many）」とは、1つの親レコードが、複数の子レコードを持つ関係のことです。

例：

1つのチーム（Team）に、複数のヒーロー（Hero）が所属する

👷 モデル定義の例

from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationship

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(default=None, foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="heroes")```

🔁 リレーションの構成ポイント

Hero（子テーブル） に team_id フィールドを定義して外部キーを張る
Team（親テーブル） は heroes リストで子の一覧を持つ
双方向の関連には Relationship(back_populates="...") を使用
- Relationship(back_populates="...") は Team, Hero の両方向にアクセスできるようにする仕組み
- これを定義すると Python コード上で同期される、以下が例
  - ただし、DBに保存されてるわけではないので、session.add, commit は必要

team = Team(name="Avengers")
hero = Hero(name="Iron Man")

team.heroes.append(hero)
# このとき、back_populates があると hero.team にも team が入る

ikepon

 2.3 多対多（Many to Many）
 💡 関係性1人のヒーローは複数のミッションに参加できる
1つのミッションには複数のヒーローが参加する
→ Hero × Mission = 多対多

 ✅ 中間テーブルを定義from typing import Optional
from sqlmodel import SQLModel, Field

class HeroMissionLink(SQLModel, table=True):
    hero_id: Optional[int] = Field(default=None, foreign_key="hero.id", primary_key=True)
    mission_id: Optional[int] = Field(default=None, foreign_key="mission.id", primary_key=True)

 ✅ Hero, Mission モデルfrom typing import List
from sqlmodel import Relationship

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    missions: List["Mission"] = Relationship(
        back_populates="heroes",
        link_model=HeroMissionLink
    )

class Mission(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    code_name: str
    heroes: List[Hero] = Relationship(
        back_populates="missions",
        link_model=HeroMissionLink
    )
link_model は、SQLModel において 多対多（Many to Many）リレーションを定義するための中間テーブル（リンクモデル）を指定するための引数です。
多対多のリレーションでは、直接2つのテーブルをつなぐことはできず、中間テーブルを経由する必要があります。

その中間テーブルを Relationship() に伝えるのが link_model の役割です。

 ✅ データを追加して関連づける例SQLModel は link_model をもとに 内部で JOIN や INSERT を自動生成します。
明示的に中間テーブルを操作しなくても、以下のような操作が可能になります：
hero = Hero(name="Spider-Boy", secret_name="SB")
mission1 = Mission(code_name="Operation Thunder")
mission2 = Mission(code_name="Operation Ice")

hero.missions.append(mission1)
hero.missions.append(mission2)

session.add(hero)
session.commit()

 ⚠️ link_model を省略すると？SQLModel では、link_model を指定しないと多対多のリレーションは構築できません。

これは SQLModel の制約であり、SQLAlchemy のように secondary="..." で省略はできません。

 ✅ データを取得するヒーローが参加しているミッション：
hero = session.get(Hero, 1)
for mission in hero.missions:
    print(mission.code_name)
ミッションに参加しているヒーロー：
mission = session.get(Mission, 1)
for hero in mission.heroes:
    print(hero.name)

 ⚠️ 補足：中間テーブルを直接使うことはほぼないHeroMissionLink は Relationship(..., link_model=...) に渡すだけでOK。

必要があれば、このモデルに追加情報（例：joined_at, role）を追加することも可能。

 ✅ まとめ

モデル
役割


Hero
複数のミッションに参加する側

Mission
複数のヒーローを含む作戦

HeroMissionLink
多対多の中間モデル（外部キー2つを持つ）

Relationship(...)
双方向のリンクを実現

🔗 公式ドキュメント: Many-to-Many Relationships

モデル	役割
`Hero`	複数のミッションに参加する側
`Mission`	複数のヒーローを含む作戦
`HeroMissionLink`	多対多の中間モデル（外部キー2つを持つ）
`Relationship(...)`	双方向のリンクを実現

ikepon

👪 2.4 多段リレーション（親 → 子 → 孫）

💡 構成イメージ

今回は Hero を中心に、次のような3階層のリレーションを構築します：

Headquarter（親）：ヒーローの本拠地
Hero（子）：各ヒーローは1つの本拠地に所属
Power（孫）：各ヒーローは複数の能力（Power）を持つ

✅ モデル定義

from typing import Optional, List
from sqlmodel import SQLModel, Field, Relationship

class Headquarter(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    location: str
    heroes: List["Hero"] = Relationship(back_populates="headquarter")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    headquarter_id: Optional[int] = Field(default=None, foreign_key="headquarter.id")
    headquarter: Optional[Headquarter] = Relationship(back_populates="heroes")
    powers: List["Power"] = Relationship(back_populates="hero")

class Power(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    hero_id: Optional[int] = Field(default=None, foreign_key="hero.id")
    hero: Optional[Hero] = Relationship(back_populates="powers")

✅ データの挿入例

hq = Headquarter(location="New York")
hero = Hero(name="Spider-Boy", headquarter=hq)
power1 = Power(name="Wall Climb", hero=hero)
power2 = Power(name="Web Swing", hero=hero)

session.add(hq)
session.add(hero)
session.add_all([power1, power2])
session.commit()

✅ リレーションの活用（データの取得）

本拠地からヒーローたちを見る：

hq = session.get(Headquarter, 1)
for hero in hq.heroes:
    print(hero.name)

ヒーローからパワーを見る：

hero = session.get(Hero, 1)
for power in hero.powers:
    print(power.name)

本拠地から孫（パワー）を見る（間接的に）：

hq = session.get(Headquarter, 1)
for hero in hq.heroes:
    for power in hero.powers:
        print(f"{hero.name} has {power.name}")

⚠️ 注意点

Relationship(back_populates=...) を必ず両方向に設定することで、Python オブジェクト間の同期が保たれます
多段であっても session.get(...) で親を取得すれば、ネストされたリストとして子・孫もたどれます

親から孫を取る方法

例としては変だが、 hq.powrs など、親から孫を直接引く方法を考える

@property で定義する

class Headquarter(SQLModel, table=True):
    # 既存項目があって、以下を追加
    @property
    def powers(self) -> List["Power"]:
        return [power for hero in self.heroes for power in hero.powers]

# 以下で取れる
hq = session.get(Headquarter, 1)
hq.powers

関連をたどる: ちょっと長いけど

class Headquarter(SQLModel, table=True):
    # 既存項目があって、以下を追加
    powers: List["Power"] = Relationship(
        back_populates="headquarter",
        sa_relationship_kwargs=dict(
            secondary="hero",  # 中間テーブルとして Hero を使う
            primaryjoin="Headquarter.id==Hero.headquarter_id",
            secondaryjoin="Hero.id==Power.hero_id",
            viewonly=True
        )
    )

class Power(SQLModel, table=True):
    # 既存項目があって、以下を追加
    headquarter: Optional[Headquarter] = Relationship(
        back_populates="powers",
        sa_relationship_kwargs=dict(
            secondary="hero",
            primaryjoin="Power.hero_id==Hero.id",
            secondaryjoin="Hero.headquarter_id==Headquarter.id",
            viewonly=True
        )
    )

secondary
- 意味：多対多や中間経由リレーションにおいて、仲介に使うテーブル（中間モデル）
- 指定するもの：中間テーブルの名前（テーブル名 or モデル名の文字列）
- 例：secondary="hero"
primaryjoin
- 意味：親モデルと中間モデルをどう結びつけるかを示す条件式（SQLのJOIN句のようなもの）
- 構文："<親モデル>.<カラム> == <中間モデル>.<カラム>"
- 例："Headquarter.id == Hero.headquarter_id"
secondaryjoin
- 意味：中間モデルと子モデル（または孫）をどう結びつけるかを示す条件式
- 構文："<中間モデル>.<カラム> == <子モデル>.<カラム>"
- 例："Hero.id == Power.hero_id"
viewonly=True
- 意味：このリレーションを読み取り専用にする
- なぜ必要？：
  - 複雑なJOINを使ったリレーションでは、SQLAlchemyが書き込み経路を判断できない
    - hq.powers.append(new_power) で commit するとエラーになる
    - -> そのため、 viewonly=True を設定して、関連を含めてレコードを作成するときは一つずつ作成する必要がある
  - viewonly=True を指定することで、ORMがこの関係を「読み取り専用」として扱う
- 必須？：多段・多対多リレーションでは原則必須
ちょっと違うけど、select, join でも取れる

from sqlmodel import select

statement = (
    select(Power)
    .join(Hero)
    .join(Headquarter)
    .where(Headquarter.id == 1)
)
powers = session.exec(statement).all()

✅ まとめ

モデル	関係
`Headquarter`	1つの本拠地に複数のヒーロー
`Hero`	本拠地に属し、複数の能力を持つ
`Power`	1人のヒーローに属する能力

多段のリレーションも SQLModel の Relationship を重ねるだけで自然に表現できる

🔗 参考: SQLModel Relationship Docs

ikepon

3. SQLModel によるデータ操作

DB接続とテーブル作成

まず、SQLModel.metadata.create_all() を使ってテーブルを作成します。

from sqlmodel import create_engine, SQLModel

# DB を指定
engine = create_engine("sqlite:///database.db")

# Model の定義からテーブルを作成する
# SQLModel.metadata: table=True で定義されているものの情報
# 既存テーブルがすでにある場合はスルーされる
# 既存テーブルの変更は反映されない -> SQLModelでは差分検出ができない
SQLModel.metadata.create_all(engine)

ローカルのテスト環境だからこの作りにしてる
- 本番なら alembic とかマイグレーションファイルで管理した方が良い

🆕 Create：レコードを追加する

データベースに新しいレコードを追加するには、Session を使います。
Session は DB への接続コンテキストでトランザクション単位で DB を操作するもの

from sqlmodel import Session

hero = Hero(name="Spider-Boy", secret_name="SB", age=16)

# Session(engine) で DB とのコネクションを作成、with で自動的に閉じるようにしてる
with Session(engine) as session:
    session.add(hero)
    session.commit()
    session.refresh(hero)  # 自動的に id などを取得

print(hero.id)  # → 自動採番された ID が表示される

📖 Read：レコードを取得する

✅ 全件取得

with Session(engine) as session:
    heroes = session.query(Hero).all()
    for hero in heroes:
        print(hero.name)

✅ 単一取得（ID）

with Session(engine) as session:
    # get(Model, ID): IDは主キー
    hero = session.get(Hero, 1)
    print(hero.name)

✅ 条件付き取得（`select()`）

基本構文

statement = select(Model).where(Model.field == 条件)
results = session.exec(statement).all()

from sqlmodel import select

with Session(engine) as session:
    statement = select(Hero).where(Hero.age >= 18)
    results = session.exec(statement)
    for hero in results:
        print(hero.name)

複数条件：AND / OR

from sqlalchemy import and_, or_

# 年齢18歳以上かつ有効なヒーロー
statement = select(Hero).where(
    and_(
        Hero.age >= 18,
        Hero.is_active == True
    )
)

# 年齢18歳未満または非アクティブ
statement = select(Hero).where(
    or_(
        Hero.age < 18,
        Hero.is_active == False
    )
)

# 年齢が18歳以上 かつ 名前が “Spider-Boy” または “Iron Man”
# 以下のように入れ子にして使える
statement = select(Hero).where(
    and_(
        Hero.age >= 18,
        or_(
            Hero.name == "Spider-Boy",
            Hero.name == "Iron Man"
        )
    )
)

並び替え（ORDER BY）

statement = select(Hero).order_by(Hero.age.desc())

最初の1件だけ取得（.first()）

statement = select(Hero).where(Hero.name == "Spider-Boy")
hero = session.exec(statement).first()

部分一致・LIKE検索

# 名前に "Spider" を含む
statement = select(Hero).where(Hero.name.contains("Spider"))

# 前方一致
statement = select(Hero).where(Hero.name.startswith("Spider"))

# 後方一致
statement = select(Hero).where(Hero.name.endswith("Boy"))

IN句（複数値に一致）

statement = select(Hero).where(
    Hero.name.in_(["Spider-Boy", "Iron Man"])

NULL チェック

# age が NULL のレコード
statement = select(Hero).where(Hero.age.is_(None))

# age が NULL ではないレコード
statement = select(Hero).where(Hero.age.is_not(None))

LIMIT / OFFSET（ページネーション）

statement = select(Hero).limit(10).offset(20)

🔄 Update：レコードを更新する

既存のレコードを変更するには、取得して値を変更し、再度 add() → commit() します。

with Session(engine) as session:
    hero = session.get(Hero, 1)
    hero.age = 30
    session.add(hero)
    session.commit()

❌ Delete：レコードを削除する

with Session(engine) as session:
    hero = session.get(Hero, 1)
    session.delete(hero)
    session.commit()

🔁 その他の操作

✅ 複数レコードの一括追加

with Session(engine) as session:
    session.add_all([Hero(name="Iron Man", secret_name="Tony"), Hero(name="Thor", secret_name="Thor")])
    session.commit()

✅ 自動コミットされない場合は `flush()` で中間反映も可能

flush() は、コミットせずに、Session内の変更を「一時的にデータベースに反映する」処理

commit() のようにトランザクションを確定はしない
でも INSERT / UPDATE / DELETE の SQL文は実行される
主キー（id）などが必要な時に使う

    session.add(hero)
    session.flush()  # ここで DB に反映されるが、commit はまだされていない

何に使う？
- トランザクションを確定させずに、一部だけを先に DB に反映させたい場面があります。
- 例：id が auto increment の場合
  - flush() を使うことで、INSERT が実行されて id が確定する
  - でも commit() してないので、あとで rollback も可能
注意点
- flush() したデータはトランザクション内の状態。commit() しないと他からは見えない
- flush() は「DBに反映 ≠ 永続化」なので、途中でエラーが出ればロールバックされる

hero = Hero(name="Thor")
session.add(hero)
session.flush()  # ここで hero.id が確定

power = Power(name="Lightning", hero_id=hero.id)
session.add(power)
session.commit()

✅ まとめ：SQLModel における基本操作

操作	メソッド	説明
Create	`add()`, `commit()`	レコードの追加
Read	`get()`, `select()`	単一 or 条件付き取得
Update	`add()`, `commit()`	値を変更して再度追加＋コミット
Delete	`delete()`, `commit()`	削除対象を指定してコミット

ikepon

 4. SQLModel の応用
 ✅ 4.1 リレーションの応用（JOINと自動読み込み）
 🔗 selectinload でリレーションをまとめて取得（N+1問題の回避）from sqlmodel import select
from sqlalchemy.orm import selectinload

statement = select(Hero).options(selectinload(Hero.powers))
heroes = session.exec(statement).all()
2025-04-19 17:51:40,755 INFO sqlalchemy.engine.Engine SELECT hero.id, hero.name, hero.secret_name, hero.age, hero.team_id, hero.headquarter_id
FROM hero
2025-04-19 17:51:40,756 INFO sqlalchemy.engine.Engine [generated in 0.00091s] ()
2025-04-19 17:51:40,761 INFO sqlalchemy.engine.Engine SELECT power.hero_id AS power_hero_id, power.id AS power_id, power.name AS power_name
FROM power
WHERE power.hero_id IN (?, ?, ?, ?, ?)
2025-04-19 17:51:40,762 INFO sqlalchemy.engine.Engine [generated in 0.00103s] (1, 2, 3, 4, 5)

selectinload を使うと、関連データ（Powerなど）を別クエリで一括取得
ORMが自動で関連づけてくれる

 🔗 JOINでリレーションを手動で結合from sqlmodel import select

statement = (
    select(Hero, Power)
    .join(Power, Power.hero_id == Hero.id)
    .where(Power.name == "Web Swing")
)
results = session.exec(statement).all()
2025-04-19 17:54:06,900 INFO sqlalchemy.engine.Engine SELECT hero.id, hero.name, hero.secret_name, hero.age, hero.team_id, hero.headquarter_id, power.id AS id_1, power.name AS name_1, power.hero_id
FROM hero JOIN power ON power.hero_id = hero.id
WHERE power.name = ?
2025-04-19 17:54:06,900 INFO sqlalchemy.engine.Engine [generated in 0.00057s] ('Web Swing',)

In [9]: results
Out[9]: [(Hero(id=1, age=None, headquarter_id=1, name='Deadpond', secret_name='Dive Wilson', team_id=1), Power(id=2, hero_id=1, name='Web Swing'))]
JOINの結果は (Hero, Power) のタプルで返る

 ✅ 4.2 中間テーブル付きの多対多（Many-to-Many）    class HeroMissionLink(SQLModel, table=True):
        hero_id: int = Field(foreign_key="hero.id", primary_key=True)
        mission_id: int = Field(foreign_key="mission.id", primary_key=True)

    class Hero(SQLModel, table=True):
        ...
        missions: List["Mission"] = Relationship(
            back_populates="heroes",
            link_model=HeroMissionLink
        )

    class Mission(SQLModel, table=True):
        ...
        heroes: List[Hero] = Relationship(
            back_populates="missions",
            link_model=HeroMissionLink
        )
多対多を自然にモデルで表現できる

link_model に属性を加えると「関係に情報を持たせる」ことも可能（例：参加日時）

 ✅ 4.3 サブクラス化と mixin（コード再利用）共通カラムを BaseModel にまとめて、継承することでDRYな定義に：
例えば、複数のテーブルに「作成日時」「更新日時」があるとします。
同じカラムを毎回コピペで書くと：
保守性が低い
バグの元
変更が面倒
➡️ 共通の親クラス（Mixin）を作って継承すれば、1箇所で定義できて再利用可能！
class TimeStamped(SQLModel):
    # default_factory は、「その場で関数を実行して初期値を作る」ための仕組みで現在時刻を入れてる
    # default=datetime.utcnow() だとモデル定義時の時刻になる、
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

# SQLModel, TimeStamped 部分は多重継承
class Hero(SQLModel, TimeStamped, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

 ✅ 4.4 カスタムプロパティ・メソッドSQLModel のクラスにPython的なロジックを追加可能：
class Hero(SQLModel, table=True):
    ...
    @property
    def age_group(self) -> str:
        return "teen" if self.age and self.age < 20 else "adult"

hero.age_group  # → "teen" or "adult"

 ✅ 4.5 自動マイグレーション対応（Alembic）
create_all() は再実行しても既存構造を変更できない
実運用では Alembic を使ってテーブル変更を管理すべき
alembic init alembic
alembic revision --autogenerate -m "add new column"
alembic upgrade head
SQLModel の構造変更をトラック＆反映できる

 ✅ 4.6 バリデーション（Pydantic の力を活かす）SQLModel は Pydantic ベースなので、入力バリデーションも標準で可能
class HeroBase(SQLModel):
    name: str = Field(min_length=2)
    age: int = Field(ge=0, le=150)
API入力チェックや、DB保存前の検証に活用できる

 ✅ 4.7 FastAPIとの統合でスキーマを分離class HeroCreate(SQLModel):
    name: str
    age: Optional[int]

class HeroRead(SQLModel):
    id: int
    name: str

@app.post("/heroes", response_model=HeroRead)
def create(hero: HeroCreate, session: Session = Depends(get_session)):
    db_hero = Hero.from_orm(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero
入力（Create）と出力（Read）をモデルで分離
APIの責務ごとに型を明示できて保守性◎

 ✅ まとめ：SQLModelの応用ポイント

応用機能
説明


selectinload
関連の一括取得でN+1問題を回避

多対多（link_model）
中間テーブルで自然な多対多モデルを構築

Mixin/継承
共通フィールドの再利用でDRYな設計

property/method
Pythonオブジェクトらしい振る舞いを追加できる

Alembic連携
マイグレーションによる変更管理

バリデーション
Pydantic の機能で入力検証も簡単

FastAPI統合
スキーマ分離で入力・出力・DBを明示的に設計可能


 5. まとめこの章では、SQLModel の基礎から応用までを一通り学んできました。

FastAPI との組み合わせや、実運用を見据えたベストプラクティスも含め、要点を以下に整理します。

 ✅ 1. SQLModel の基礎SQLAlchemy + Pydantic の融合モデル

SQLModel を継承して table=True を指定することで、テーブル定義が可能
型安全、入力検証、スキーマ定義を1つで行える

 ✅ 2. モデル定義とリレーション
Field(...) でカラム属性を指定（例：primary_key=True, foreign_key=...）

Relationship(...) でリレーション定義（1対多・多対1・多対多）

 多対多
link_model= を指定して中間テーブルを使う
中間モデルに属性（例：参加日時など）を追加可能

 多段（親→子→孫）
Relationship を段階的に重ねる

@property を使えば祖先→孫アクセスも可能（ただし viewonly）

 ✅ 3. データ操作（CRUD）
Session で DB 操作を管理
基本操作：

.add() → .commit() → .refresh()

.get() / .delete() / select(...) / .exec()


 条件付き取得（select）
where(), and_(), or_() で絞り込み

order_by(), limit(), offset() で並び順やページング

 ✅ 4. 応用テクニック
 🧱 サブクラス化・Mixin
TimestampMixin, SoftDeleteMixin などで共通項目を切り出す

default_factory を使って日時や UUID を動的に生成

 🔁 updated_at の自動更新SQLAlchemy の event.listen() を使って before_update フックを仕込む

SQLModel 全体に適用したい場合は propagate=True

 🧠 スキーマ分離（FastAPI）入力用：HeroCreate
出力用：HeroRead
DB用：Hero（SQLModel + table）

 🔄 selectinload（リレーション一括取得）
select(...).options(selectinload(...)) で N+1 問題を回避

 🔧 Alembicとの統合
create_all() は試作用
本番では alembic revision --autogenerate → upgrade を使う

 ✅ ベストプラクティスまとめ

やること
方法・ポイント


モデルを定義する

SQLModel, Field, Relationship を使う

複数処理をまとめて管理

with Session(engine) as session: を使い回す

多対多を定義する

link_model= を使って中間テーブルを指定


updated_at を自動更新

event.listen(SQLModel, "before_update", ...) を使う

モデルの共通化

Mixin クラスを作って継承

API でスキーマを分離する

HeroCreate, HeroRead, Hero で目的別に設計

実運用の構造管理
Alembic でマイグレーション履歴を管理

📌 SQLModel を使うことで、Pythonらしいコードでデータベースとやり取りでき、

FastAPIと組み合わせれば、高速・型安全・自動ドキュメント付きAPIを素早く構築できます。

 その他調査事項FastAPI などで 1 リクエスト内で複数DB処理を行う場合、1つのSessionで処理するのと複数のSessionで処理するのはどっちが良い？
結論：基本は 1リクエスト = 1Session
API処理では、「1つのリクエストの中で、1つの Session を使い回す」 のが一般的で、ベストプラクティス

理由（なぜ1セッションでまとめるのが良いのか）
トランザクション管理が簡単になる
session.commit() までの間は「未確定の変更」がある
一連の操作（insert → update → deleteなど）を 一つのトランザクションとして扱える
途中でエラーが起きても session.rollback() で一括リセットできる

接続コストが抑えられる
SQLiteやPostgreSQLでは セッション = DB接続 のことが多い
毎回 with Session(...) を使うと コネクションを毎回開閉することになる
→ オーバーヘッドが増える & パフォーマンスが低下


ORMの状態管理が正しく動く
SQLAlchemy（SQLModelのベース）は、Session内でオブジェクトの変更を追跡（Unit of Work） します
セッションを分けると、同じオブジェクトを何度も読み直すことになり、状態不整合の原因になります


多重継承のルール

class Hero(SQLModel, TimeStamped, table=True): は順序が大事
この順序だと、SQLModel → TimeStamped の順に MRO（メソッド解決順序）が働く
一般的には、Mixin クラス（TimeStampedなど）は最後に書くのが慣習です

（SQLModel は本体、TimeStamped は追加機能というイメージ）

🔄 updated_at を自動更新する方法（SQLModel + SQLAlchemy）
モデルを更新するたびに updated_at カラムに 現在の時刻を自動で記録
手動で毎回 obj.updated_at = datetime.utcnow() を書かなくて済むようにしたい
注意点

before_update は「実際に UPDATE が発行される時」にのみ呼ばれる
→ 値を変えないと呼ばれない（差分がないとスキップされる）
自動で更新されるのは session.commit() 時

flush() のタイミングでも呼ばれる

from sqlalchemy import event
from sqlalchemy.orm import Mapper
from sqlmodel import SQLModel

def auto_update_updated_at(mapper: Mapper, connection, target):
    from datetime import datetime
    target.updated_at = datetime.utcnow()

event.listen(SQLModel, "before_update", auto_update_updated_at, propagate=True)


部分
内容


event.listen()
SQLAlchemy にイベント（hook）を登録する

"before_update"
UPDATE 文が実行される直前に呼ばれる

propagate=True
SQLModel を継承したすべてのクラスにも適用する

target
更新されるインスタンス（例：Heroのインスタンス）

🔗 公式ドキュメント: https://sqlmodel.tiangolo.com/

応用機能	説明
selectinload	関連の一括取得でN+1問題を回避
多対多（link_model）	中間テーブルで自然な多対多モデルを構築
Mixin/継承	共通フィールドの再利用でDRYな設計
property/method	Pythonオブジェクトらしい振る舞いを追加できる
Alembic連携	マイグレーションによる変更管理
バリデーション	Pydantic の機能で入力検証も簡単
FastAPI統合	スキーマ分離で入力・出力・DBを明示的に設計可能

やること	方法・ポイント
モデルを定義する	`SQLModel`, `Field`, `Relationship` を使う
複数処理をまとめて管理	`with Session(engine) as session:` を使い回す
多対多を定義する	`link_model=` を使って中間テーブルを指定
`updated_at` を自動更新	`event.listen(SQLModel, "before_update", ...)` を使う
モデルの共通化	`Mixin` クラスを作って継承
API でスキーマを分離する	`HeroCreate`, `HeroRead`, `Hero` で目的別に設計
実運用の構造管理	Alembic でマイグレーション履歴を管理

部分	内容
`event.listen()`	SQLAlchemy にイベント（hook）を登録する
`"before_update"`	UPDATE 文が実行される直前に呼ばれる
`propagate=True`	SQLModel を継承したすべてのクラスにも適用する
`target`	更新されるインスタンス（例：Heroのインスタンス）

とりあえす動かしてみる

データ作成

SQLModel完全入門：基礎から複雑なリレーションまで

1. SQLModelの基礎

特徴

2. SQLModel による定義

2.1 1テーブルの定義

2.2 1 対 多（One to Many）

👷 モデル定義の例

🔁 リレーションの構成ポイント

2.3 多対多（Many to Many）

💡 関係性

✅ 中間テーブルを定義

✅ Hero, Mission モデル

✅ データを追加して関連づける例

⚠️ link_model を省略すると？

✅ データを取得する

⚠️ 補足：中間テーブルを直接使うことはほぼない

✅ まとめ

👪 2.4 多段リレーション（親 → 子 → 孫）

💡 構成イメージ

✅ モデル定義

✅ データの挿入例

✅ リレーションの活用（データの取得）

⚠️ 注意点

親から孫を取る方法

✅ まとめ

3. SQLModel によるデータ操作

DB接続とテーブル作成

🆕 Create：レコードを追加する

📖 Read：レコードを取得する

✅ 全件取得

✅ 単一取得（ID）

✅ 条件付き取得（select()）

🔄 Update：レコードを更新する

❌ Delete：レコードを削除する

🔁 その他の操作

✅ 複数レコードの一括追加

✅ 自動コミットされない場合は flush() で中間反映も可能

✅ まとめ：SQLModel における基本操作

4. SQLModel の応用

✅ 4.1 リレーションの応用（JOINと自動読み込み）

🔗 selectinload でリレーションをまとめて取得（N+1問題の回避）

🔗 JOINでリレーションを手動で結合

✅ 4.2 中間テーブル付きの多対多（Many-to-Many）

✅ 4.3 サブクラス化と mixin（コード再利用）

✅ 4.4 カスタムプロパティ・メソッド

✅ 4.5 自動マイグレーション対応（Alembic）

✅ 4.6 バリデーション（Pydantic の力を活かす）

✅ 4.7 FastAPIとの統合でスキーマを分離

✅ まとめ：SQLModelの応用ポイント

5. まとめ

✅ 1. SQLModel の基礎

✅ 2. モデル定義とリレーション

多対多

多段（親→子→孫）

✅ 3. データ操作（CRUD）

条件付き取得（select）

✅ 4. 応用テクニック

🧱 サブクラス化・Mixin

🔁 updated_at の自動更新

🧠 スキーマ分離（FastAPI）

🔄 selectinload（リレーション一括取得）

🔧 Alembicとの統合

✅ ベストプラクティスまとめ

その他調査事項

2.2 1 対多（One to Many）

⚠️ `link_model` を省略すると？

✅ 条件付き取得（`select()`）

✅ 自動コミットされない場合は `flush()` で中間反映も可能