2
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@miso_taku

FastAPI×SQLModel入門

2
Posted at

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" }

image.png

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

image.png

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にアクセス

image.png

  • 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へ
2
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?