FastAPI×SQLModel入門

1. なぜ SQLModel？（超要約）

• SQLModel = Pydantic + SQLAlchemy の良いとこ取り
• モデル1つで バリデーション（スキーマ） と DBテーブル（ORM） を兼用
• 型安全に書けて、SQLAlchemy の強いクエリ機構もそのまま使える

メモ: SQLModel の最新版は Pydantic v2 に対応。もし v1 系を使っている環境なら、本文の model_dump() を dict() に読み替えればOK。

---

2. 検証環境とセットアップ

環境

• Windows / PowerShell
• Python 3.11 以上（想定）

パッケージ

• fastapi, sqlmodel, uvicorn[standard]

初期化

mkdir fastapi-sqlmodel-demo
cd fastapi-sqlmodel-demo
uv init
uv add fastapi sqlmodel "uvicorn[standard]"

---

3. Step 0: FastAPI の最小起動（/ping）

app.py

from fastapi import FastAPI

app = FastAPI()

# ヘルスチェック用の最小エンドポイント
@app.get("/ping")
def ping():
    # FastAPI は dict を返すと JSON に変換して返してくれる
    return {"status": "ok"}

起動:

uv run uvicorn app:app --reload

• ブラウザでhttp://127.0.0.1:8000/pingにアクセス → { "status": "ok" }

• http://127.0.0.1:8000/docs → 自動ドキュメント

---

4. Step 1: 単一ファイルで CRUD を完成（SQLModel × SQLite）

まずは 1ファイルで全体像を掴む。
※初心者にもわかりやすいようにコード内に解説コメントを多めに書いています。

app.py を丸ごと置き換え：

# app.py（単一ファイルで CRUD まで）
from typing import Optional
from fastapi import FastAPI, Depends, HTTPException, status
from sqlmodel import SQLModel, Field, Session, create_engine, select

# --- DB接続の設定 ---
# SQLite の DB ファイル。相対パスなので「uv を実行したカレントディレクトリ」に作成されます。
# 学習時は echo=True で発行される SQL を観察できるようにしています（実運用では False 推奨）。
sqlite_url = "sqlite:///app.db"
engine = create_engine(sqlite_url, echo=True)

# アプリ起動時にテーブルを作成（簡易運用）
# 本格運用では Alembic 等のマイグレーションを使うのが定石です。
def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)

# リクエストごとに新しい Session を提供する依存関係
# `Depends(get_session)` と書いたエンドポイントにセッションが注入され、処理後に必ずクローズされます。
def get_session():
    with Session(engine) as session:
        yield session

# --- モデル定義（SQLModel） ---
# TaskBase は共通フィールドの“型”と“制約”をまとめるための基底クラス
class TaskBase(SQLModel):
    # index=True で検索に使うカラムにインデックスを張る（SQLite でも有効）
    # min_length/max_length は Pydantic 側のバリデーション（DB 側の長さ制約ではない点に注意）
    title: str = Field(index=True, min_length=1, max_length=200)
    # 既定値 False（未指定なら False が入る）
    done: bool = False

# table=True を付けると「テーブル（ORM）」として扱われる
class Task(TaskBase, table=True):
    # 主キー。Auto Increment。INSERT 後に採番される
    id: Optional[int] = Field(default=None, primary_key=True)

# 入力（作成時）用スキーマ。TaskBase をそのまま使う
class TaskCreate(TaskBase):
    pass

# 出力（読み取り）用スキーマ。API レスポンスに id を含めたいので別定義
class TaskRead(TaskBase):
    id: int

# 更新（部分更新）用スキーマ。すべて任意にして PATCH に対応
class TaskUpdate(SQLModel):
    title: Optional[str] = None
    done: Optional[bool] = None

# --- FastAPI 本体 ---
app = FastAPI()

# アプリ起動時に一度だけテーブルを作成
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# 動作確認用
@app.get("/ping")
def ping():
    return {"status": "ok"}

# --- CRUD エンドポイント ---
# Create: 1件作成
@app.post("/tasks", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: Session = Depends(get_session)):
    # Pydantic v2: model_dump() / v1: dict()
    task = Task(**payload.model_dump())
    session.add(task)      # まずはトランザクションに追加
    session.commit()       # ここで実際に INSERT 発行＆確定
    session.refresh(task)  # DB 側で決まった値（id など）を Python オブジェクトに反映
    return task

# Read: 一覧取得
@app.get("/tasks", response_model=list[TaskRead])
def list_tasks(session: Session = Depends(get_session)):
    # select(Task) で SELECT * FROM task 相当のクエリを構築
    return session.exec(select(Task)).all()

# Read: 1件取得（主キー）
@app.get("/tasks/{task_id}", response_model=TaskRead)
def get_task(task_id: int, session: Session = Depends(get_session)):
    # 主キー検索の最短ルート。見つからなければ None
    task = session.get(Task, task_id)
    if not task:
        # FastAPI 標準の例外 → 適切な HTTP ステータス/JSON に変換される
        raise HTTPException(status_code=404, detail="Task not found")
    return task

# Update: 部分更新（PATCH）
@app.patch("/tasks/{task_id}", response_model=TaskRead)
def update_task(task_id: int, patch: TaskUpdate, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    # 送られてきた項目だけを取り出す。未指定フィールドの意図しない上書きを防ぐ。
    changes = patch.model_dump(exclude_unset=True)  # v1: dict(exclude_unset=True)
    for k, v in changes.items():
        setattr(task, k, v)
    session.add(task)
    session.commit()      # UPDATE を確定
    session.refresh(task) # 反映後の最新状態を再取得
    return task

# Delete: 削除
@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    session.delete(task)
    session.commit()  # DELETE を確定

実行:

uv run uvicorn app:app --reload

• ブラウザで http://127.0.0.1:8000/docsにアクセス

• POST /tasks から Try it out → POST に {"title":"first task"} を入れて Execute で送信

• 続けて GET /tasks で Execute 一覧が返るのを確認

• 同様に PACH /tasks , DELETE /tasks も確認

ポイント

• Depends(get_session) でセッションをリクエストごとに供給
• commit()→refresh() で DB 確定値（採番 ID 等）を反映
• exclude_unset=True で 部分更新 を安全に（未指定フィールドの上書き防止）
• 404/201/204 など 適切なHTTPステータス を返す

CRUDの動き（解説）

POST /tasks（作成）

• 送信した title で新しい行を作成し、commit() で保存。refresh() で採番IDがオブジェクトに戻ります。
• 例：

{"title": "first task"}

{"id": 1, "title": "first task", "done": false}

GET /tasks（一覧）

• 全ての Task を配列で返します。

[
  {"id": 1, "title": "first task", "done": false}
]

GET /tasks/{id}（1件取得）

• session.get(Task, id) で主キー検索。無いときは 404 を返します。

PATCH /tasks/{id}（部分更新）

• 送られた項目だけを更新（exclude_unset=True）。

{"done": true}

{"id": 1, "title": "first task", "done": true}

DELETE /tasks/{id}（削除）

• 成功すると 204 No Content（本文なし）。

補足：Depends(get_session) で各リクエストに新しい DB セッションが渡され、処理後に自動でクローズされます。

---

5. Step 2: ファイル分割（db / models / schemas / main）

単一ファイルから 責務ごとの分割 へ。
まずはパッケージ構成に分割して「起動できる」状態まで作ります（この段階では /tasks はまだ追加しません）。そのあとで Repository と Router を足します。

app/
  __init__.py
  main.py        # FastAPI本体（/ping など）
  db.py          # Engine, Session依存性, 初期化
  models.py      # SQLModel のテーブル定義
  schemas.py     # 入出力用スキーマ（Create/Read/Update）

app/db.py

from sqlmodel import SQLModel, Session, create_engine

# SQLite の DB ファイル。相対パスは実行カレントに依存します。
# 学習中は echo=True で実行される SQL を観察。実運用では False 推奨。
sqlite_url = "sqlite:///app.db"
engine = create_engine(sqlite_url, echo=True)

# アプリ起動時にテーブルを一括作成（簡易運用向け）
# 変更が増えてきたら Alembic などでマイグレーション管理に切替えましょう。
def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)

# FastAPI の Depends で使うセッション供給関数。
# with ブロックにより、処理の成否に関わらず確実にクローズされます。
def get_session():
    with Session(engine) as session:
        yield session

app/models.py

from typing import Optional
from sqlmodel import SQLModel, Field

# テーブル定義（ORM）
class Task(SQLModel, table=True):
    # 主キー。None のまま add→commit すると DB 側で採番される
    id: Optional[int] = Field(default=None, primary_key=True)
    # 検索に使う title にインデックスを付与。
    # min_length/max_length は Pydantic の入力バリデーション（DB の制約ではない点に注意）。
    title: str = Field(index=True, min_length=1, max_length=200)
    # 完了フラグ（既定 False）
    done: bool = False

app/schemas.py

from typing import Optional
from sqlmodel import SQLModel

# 入出力用スキーマを役割ごとに分けて、API の I/O を明確にする
class TaskBase(SQLModel):
    # 入力時のバリデーションも兼ねる
    title: str
    done: bool = False

# 作成時の入力（title 必須 / done 任意）
class TaskCreate(TaskBase):
    pass

# レスポンス用（id を含める）
class TaskRead(TaskBase):
    id: int

# 部分更新用。すべて Optional にして PATCH を安全に扱う
class TaskUpdate(SQLModel):
    title: Optional[str] = None
    done: Optional[bool] = None

app/main.py

from fastapi import FastAPI
from .db import create_db_and_tables

# アプリケーション本体
app = FastAPI()

# 起動時に一度だけテーブル作成（簡易運用）
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# 動作確認用の最小エンドポイント
@app.get("/ping")
def ping():
    return {"status": "ok"}

起動：

uv run uvicorn app.main:app --reload

---

6. Step 3: Repository + Router でレイヤ分割（DDD的な責務分離）

HTTP の処理（Router）と DB 操作（Repository）を分けると、責務が明確になりテストしやすくなります。ここではページングなどの発展機能は入れず、まずは最小の CRUD を分離します。

構成

app/
  repositories/
    __init__.py
    tasks.py
  routers/
    __init__.py
    tasks.py
  db.py
  models.py
  schemas.py
  main.py

app/repositories/tasks.py

from typing import Optional
from sqlmodel import Session, select
from ..models import Task
from ..schemas import TaskCreate, TaskUpdate

# Repository 層：DB 操作だけを集約（HTTP を一切知らない）
def create(session: Session, data: TaskCreate) -> Task:
    task = Task(**data.model_dump())  # Pydantic v1 は data.dict()
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

# 最小の一覧：まずは全件（必要になったら WHERE や ORDER BY を追加）
def list_all(session: Session) -> list[Task]:
    return session.exec(select(Task)).all()

# 主キー取得
def get_by_id(session: Session, task_id: int) -> Optional[Task]:
    return session.get(Task, task_id)

# 部分更新（未指定フィールドの上書きを避ける）
def update(session: Session, task_id: int, patch: TaskUpdate) -> Optional[Task]:
    task = session.get(Task, task_id)
    if not task:
        return None
    for k, v in patch.model_dump(exclude_unset=True).items():  # v1: dict(exclude_unset=True)
        setattr(task, k, v)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

# 削除
def delete(session: Session, task_id: int) -> bool:
    task = session.get(Task, task_id)
    if not task:
        return False
    session.delete(task)
    session.commit()
    return True

app/routers/tasks.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session
from ..db import get_session
from ..schemas import TaskCreate, TaskRead, TaskUpdate
from ..repositories import tasks as repo

# Router 層：HTTP の入出力とステータスコードの責務を担当
router = APIRouter(prefix="/tasks", tags=["tasks"])

@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: Session = Depends(get_session)):
    return repo.create(session, payload)

@router.get("", response_model=list[TaskRead])
def list_tasks(session: Session = Depends(get_session)):
    return repo.list_all(session)

@router.get("/{task_id}", response_model=TaskRead)
def get_task(task_id: int, session: Session = Depends(get_session)):
    task = repo.get_by_id(session, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@router.patch("/{task_id}", response_model=TaskRead)
def update_task(task_id: int, patch: TaskUpdate, session: Session = Depends(get_session)):
    task = repo.update(session, task_id, patch)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, session: Session = Depends(get_session)):
    ok = repo.delete(session, task_id)
    if not ok:
        raise HTTPException(status_code=404, detail="Task not found")

app/main.py（ルーターを組み込む）

from fastapi import FastAPI
from .db import create_db_and_tables
from .routers.tasks import router as tasks_router

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

@app.get("/ping")
def ping():
    return {"status": "ok"}

# 追加：/tasks ルーターを有効化
app.include_router(tasks_router)

ポイント：repositories/ と routers/ に __init__.py を置く（Python のパッケージとして認識させる）。

---

7. よくあるハマりどころ

• model_dump() が無い ⇒ Pydantic v1 系。obj.dict()／dict(exclude_unset=True) を使う
• SQLite ファイルが見つからない ⇒ sqlite:///相対パス は「実行カレント」に依存。uv run をプロジェクト直下で
• ログがうるさい ⇒ create_engine(..., echo=False) に変更
• CORS ⇒ フロントと別ドメインなら fastapi.middleware.cors で許可設定

---

8. 次の一歩（拡張アイデア）

• ページング: (limit/offset)、一覧の強化: 並び替え（ORDER BY）、検索（title 部分一致）、合計件数（total）
• モデルの堅牢化: ユニーク制約、created_at/updated_at、IntegrityError ハンドリング
• リレーション: Project–Task（1対多）を定義し、/projects/{id}/tasks を追加

9. まとめ

• 単一ファイル CRUD で挙動を理解 → 責務分割 で拡張性UP
• SQLModel は Pydantic の型安全 × SQLAlchemy の表現力で 実装スピードと保守性 を両立
• この土台に、検索/並び替え/認証/リレーション等を積み上げれば実用APIへ