FastAPIにDBを載せる本番ガイド：SQLModel（Pydantic×SQLAlchemy統合）でCRUD・複数モデル境界・リレーション・非同期Alembic

from sqlmodel import Field, SQLModel

# このクラス1つが「DBテーブル」かつ「Pydanticモデル」。
# table=True がDBテーブルとしての顔を有効にする（後述：データ専用モデルは table を付けない）。
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None

from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    # id は DB が自動採番する。だから「未確定（None）でも作れる」が、DB上は主キー。
    id: int | None = Field(default=None, primary_key=True)
    # index=True で検索高速化のインデックスを張る（name で頻繁に絞るなら付ける）。
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str

from sqlmodel import SQLModel, create_engine

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

# SQLite固有の注意：FastAPIは1リクエストを複数スレッドで触り得るので、
# check_same_thread=False で「別スレッドからの利用」を許可する。
# ※これはSQLite限定の事情。PostgreSQLでは不要（第7章で差し替える）。
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)

def create_db_and_tables() -> None:
    # 定義済みの全テーブルを「無ければ作る」。学習・プロトタイプには最適。
    SQLModel.metadata.create_all(engine)

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()   # 起動時に一度（本番ではここはAlembicに置き換える）
    yield                    # ← アプリ稼働中
    # 終了時の後始末が要ればここに書く

app = FastAPI(lifespan=lifespan)

from typing import Annotated
from fastapi import Depends
from sqlmodel import Session

def get_session():
    # with でセッションを開き、yield で渡し、ハンドラ終了後に自動でクローズ。
    # 1リクエスト = 1セッションを保証する（接続リークを構造的に防ぐ）。
    with Session(engine) as session:
        yield session

# 公式推奨：型エイリアスに畳んで全エンドポイントで再利用する（DRY）。
SessionDep = Annotated[Session, Depends(get_session)]

from fastapi import FastAPI

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)       # セッションに追加（まだDBには書かれていない）
    session.commit()        # ここでINSERTが確定する（トランザクションをコミット）
    session.refresh(hero)   # DBが採番した id 等を、オブジェクトに反映して読み戻す
    return hero

from fastapi import HTTPException

@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
    # 主キーでの取得は session.get が最短（select を書く必要が無い）。
    hero = session.get(Hero, hero_id)
    if not hero:
        # 見つからないなら404。境界で必ず存在チェックする。
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

from fastapi import Query
from sqlmodel import select

@app.get("/heroes/")
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    # 上限を Query(le=100) で固定する。無制限の limit は「全件取得DoS」になり得る。
    limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

from sqlmodel import Field, SQLModel

# 共通の土台。table を付けない＝ただのデータモデル（Pydantic相当）。
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

# テーブルモデルはこれ「だけ」。table=True を持つのは1つに保つ。
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str        # 機密。公開モデルには載せない。

# 入力スキーマ：id を持たない（クライアントに id を送らせない＝注入を防ぐ）。
class HeroCreate(HeroBase):
    secret_name: str        # 作成時は必須。だが公開レスポンスには出さない。

# 公開スキーマ：id は必須（DBから来た以上 id は必ずある）。secret_name は無い。
class HeroPublic(HeroBase):
    id: int

@app.post("/heroes/", response_model=HeroPublic)   # 出力は HeroPublic に絞る（機密を漏らさない）
def create_hero(hero: HeroCreate, session: SessionDep):  # 入力は HeroCreate（id を受けない）
    # 入力スキーマ → テーブルモデルへ変換。バリデーション済みの値だけが通る。
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero        # Hero を返しても response_model=HeroPublic で secret_name は除去される

@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(session: SessionDep, offset: int = 0,
                limit: Annotated[int, Query(le=100)] = 100):
    return session.exec(select(Hero).offset(offset).limit(limit)).all()

# 更新は「一部だけ送る」のが普通。だから全フィールドを任意（Noneデフォルト）にする。
# 公式注記：全フィールドが任意なので HeroBase からは継承できない（HeroBaseは name 必須）。
class HeroUpdate(SQLModel):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    # exclude_unset=True が肝：「クライアントが実際に送ったフィールドだけ」を dict 化する。
    # 送らなかったフィールドは dict に入らない＝既存値を維持する（誤った上書きを防ぐ）。
    hero_data = hero.model_dump(exclude_unset=True)
    # sqlmodel_update：dict の値だけをオブジェクトに適用する（SQLModel 専用の安全な更新）。
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

from sqlmodel import Field, Relationship, SQLModel

class TeamBase(SQLModel):
    name: str = Field(index=True)

class Team(TeamBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    # 1チームに複数ヒーロー。back_populates で Hero.team と双方向に結ぶ。
    heroes: list["Hero"] = Relationship(back_populates="team")

class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    # 外部キー：DBレベルで team.id を参照する。これが整合性の土台。
    team_id: int | None = Field(default=None, foreign_key="team.id")

class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str
    # 多側から1側へ。back_populates="heroes" で Team.heroes と対応させる。
    team: Team | None = Relationship(back_populates="heroes")

class TeamPublic(TeamBase):
    id: int

class HeroPublic(HeroBase):
    id: int

# ヒーロー＋所属チームを返す。team は TeamPublic（…WithHeroes ではない）にして循環を断つ。
class HeroPublicWithTeam(HeroPublic):
    team: TeamPublic | None = None

# チーム＋所属ヒーロー一覧。heroes は HeroPublic（…WithTeam ではない）にして循環を断つ。
class TeamPublicWithHeroes(TeamPublic):
    heroes: list[HeroPublic] = []

@app.get("/heroes/{hero_id}", response_model=HeroPublicWithTeam)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero    # team も含めて HeroPublicWithTeam の形で返る

import os
from sqlmodel import create_engine

# 接続URLは必ず環境変数から。コードにもgitにも資格情報を置かない。
DATABASE_URL = os.environ["DATABASE_URL"]   # 例: postgresql+psycopg://user:pass@host:5432/db

engine = create_engine(
    DATABASE_URL,
    pool_size=10,           # 常時保持する接続数（CPU/同時実行に合わせる）
    max_overflow=20,        # 一時的に超過できる上限
    pool_pre_ping=True,     # 死んだ接続を使う前に軽く疎通確認（切断後の事故を防ぐ）
    pool_recycle=1800,      # 一定時間で接続を作り直す（DB側のidleタイムアウト対策）
)
# 注意：PostgreSQL では SQLite 用の connect_args={"check_same_thread": False} は不要。

from collections.abc import AsyncGenerator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlmodel import select

# 非同期エンジン。ドライバは asyncpg（URLの +asyncpg に注目）。
async_engine = create_async_engine(
    os.environ["DATABASE_URL"],   # 例: postgresql+asyncpg://user:pass@host/db
    pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    # 非同期でも yield 依存の形は同じ。async with で確実にクローズする。
    async with AsyncSessionLocal() as session:
        yield session

AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]

@app.get("/heroes/", response_model=list[HeroPublic])
async def read_heroes(session: AsyncSessionDep):   # 非同期ハンドラ
    # 同期版の session.exec(...) は使えない。await session.execute(...) を使う。
    result = await session.execute(select(Hero).limit(100))
    return result.scalars().all()

# 初期化（一度だけ）。以後はモデルの変更を検出してマイグレーションを自動生成できる。
alembic init alembic
alembic revision --autogenerate -m "create hero and team tables"
alembic upgrade head     # 本番への適用はこのコマンド（CI/CDから流す）

from sqlalchemy.orm import selectinload

@app.get("/heroes/", response_model=list[HeroPublicWithTeam])
async def read_heroes_with_team(session: AsyncSessionDep):
    # selectinload で「関連 team を別の1クエリでまとめて取る」。N+1 が 2クエリに畳まれる。
    result = await session.execute(
        select(Hero).options(selectinload(Hero.team)).limit(100)
    )
    return result.scalars().all()

# 複数の書き込みを1トランザクションに束ねる：途中で失敗したら全体がロールバックされる。
def transfer_hero(session: Session, hero_id: int, new_team_id: int) -> None:
    hero = session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero.team_id = new_team_id
    # ここで他の書き込みを足しても、commit までは確定しない。
    session.add(hero)
    session.commit()         # ここで初めて全変更が一括確定する（途中例外なら何も残らない）

from sqlalchemy.dialects.postgresql import insert as pg_insert

def upsert_hero(session: Session, name: str, secret_name: str) -> None:
    # name を一意キーとして、衝突したら更新する（ON CONFLICT DO UPDATE）。
    # リトライや二重送信で重複行を作らない＝冪等。
    stmt = pg_insert(Hero).values(name=name, secret_name=secret_name)
    stmt = stmt.on_conflict_do_update(
        index_elements=["name"],                       # 一意制約のある列
        set_={"secret_name": secret_name},
    )
    session.execute(stmt)
    session.commit()

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from app.main import app, get_session

@pytest.fixture(name="session")
def session_fixture():
    # テスト専用：インメモリSQLite。StaticPool で「同じ接続」を共有し、テーブルを保持する。
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)     # テストではこれでよい（本番はAlembic）
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    # 本番の get_session を、テスト用セッションを返す関数で上書きする。
    def get_session_override():
        return session
    app.dependency_overrides[get_session] = get_session_override
    yield TestClient(app)
    app.dependency_overrides.clear()         # テスト間の汚染を防ぐため必ずクリア

def test_create_hero(client):
    res = client.post("/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"})
    assert res.status_code == 200
    data = res.json()
    assert data["name"] == "Deadpond"
    assert "id" in data
    assert "secret_name" not in data          # 公開モデルが機密を漏らさないことを検証

@pytest.fixture(name="session")
def session_fixture():
    # 外側のトランザクションを開始し、テスト終了時に rollback で全変更を捨てる。
    connection = engine.connect()
    transaction = connection.begin()
    with Session(bind=connection) as session:
        yield session
    transaction.rollback()      # コミットされた分も含めて巻き戻す（テスト間が完全独立）
    connection.close()