はじめに

前回までで、FastAPIの基本的なAPIの書き方を学びました。
でも今まで作ってきたAPIには「データを保存する場所」がありませんでした。

今回は SQLite をデータベースに使い、実際にデータを永続化する「ユーザー管理API」を作ります。

さらに、ORM（Pythonでテーブルを操作するライブラリ）として SQLAlchemy と SQLModel の両方で実装を紹介します。

この記事で作るもの
ユーザーの作成・一覧取得・1件取得・更新・削除ができる REST API
（GitHubにコードを上げてポートフォリオにできる内容です）

1. ORMとは？

ORM（Object-Relational Mapper）は、SQLを直接書かずにPythonのコードでデータベースを操作できるライブラリです。

# SQLを直接書く場合
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))

# ORM（SQLAlchemy）を使う場合
session.get(User, user_id)

ORM を使うと：

SQL を覚えなくても Python だけで DB 操作できる
テーブル構造をクラスで表現できる（型補完が効く）
DB の種類（SQLite/PostgreSQL など）を切り替えやすい

2. SQLAlchemy vs SQLModel

	SQLAlchemy	SQLModel
作者	Mike Bayer	Sebastián Ramírez（FastAPI作者）
歴史	2006年〜（業界標準）	2021年〜（比較的新しい）
特徴	機能豊富・実務で広く使われる	Pydantic + SQLAlchemy を統合
モデル定義	DB用クラスとPydanticクラスを別々に書く	1つのクラスで両方を兼ねる
学習コスト	やや高い	低い（FastAPIと相性が良い）
実務採用	非常に多い	増えてきている

初心者にはSQLModelが書きやすく、実務ではSQLAlchemyを見る機会が多いです。
この記事では両方紹介します。

3. プロジェクト構成

my_api/
├── main.py          # FastAPIアプリ本体
├── database.py      # DB接続設定
├── models.py        # テーブル定義（ORMモデル）
├── schemas.py       # リクエスト/レスポンスのPydanticモデル
└── crud.py          # DB操作の関数群

💡 なぜファイルを分けるの？
全部 main.py に書いても動きますが、規模が大きくなると読みづらくなります。
役割ごとにファイルを分けておくと、実務でもそのまま使えるスタイルになります。

4. SQLAlchemy で実装する

インストール

pip install fastapi uvicorn sqlalchemy

database.py ── DB接続設定

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# SQLiteファイルの場所（同じディレクトリに app.db が作られる）
DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite特有の設定
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

# FastAPIのDependency Injection用
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

💡 sqlite:///./app.db のスラッシュが3本？
初見では /// の意味がわかりづらいので分解して説明します。
sqlite  ://  ./app.db
^^^^^^  ^^^  ^^^^^^^^
スキーム 区切り ファイルパス（. = カレントディレクトリ）
:// はURLの固定の区切り記号で、その後ろがファイルパスです。
./app.db の先頭 . が「現在のディレクトリ」を意味するため、スラッシュが3本に見えます。

用途別に書き分けるとこうなります：
# 相対パス（uvicornを起動したディレクトリにapp.dbが作られる）
"sqlite:///./app.db"      # ./ = カレントディレクトリ（推奨）
"sqlite:///app.db"        # 同じ意味（./は省略可）

# 絶対パス（場所を固定したいとき）── スラッシュが4本になる
"sqlite:////home/user/myapp/app.db"

# メモリ上だけ（テスト用・サーバー終了と同時に消える）
"sqlite:///:memory:"
開発中は相対パス、本番や場所を固定したいときは絶対パスを使うのが一般的です。

models.py ── テーブル定義

from sqlalchemy import Column, Integer, String, Boolean
from database import Base

class User(Base):
    __tablename__ = "users"

    id       = Column(Integer, primary_key=True, index=True)
    name     = Column(String, nullable=False)
    email    = Column(String, unique=True, index=True, nullable=False)
    is_active = Column(Boolean, default=True)

schemas.py ── リクエスト / レスポンスの型定義

from pydantic import BaseModel, Field, EmailStr
from typing import Optional

# 作成時のリクエストボディ
class UserCreate(BaseModel):
    name:  str   = Field(min_length=1, max_length=50)
    email: str   = Field(pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")

# 更新時のリクエストボディ（全フィールド省略可）
class UserUpdate(BaseModel):
    name:      Optional[str]  = Field(None, min_length=1, max_length=50)
    is_active: Optional[bool] = None

# レスポンス用（DBのidも含む）
class UserResponse(BaseModel):
    id:        int
    name:      str
    email:     str
    is_active: bool

    model_config = {"from_attributes": True}  # SQLAlchemyオブジェクトをそのまま変換できる

💡 model_config = {"from_attributes": True} とは？
SQLAlchemyのモデルオブジェクト（Userインスタンス）をPydanticモデルに変換するための設定です。
これがないと UserResponse に変換するときにエラーになります。

crud.py ── DB操作の関数

from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate, UserUpdate

def get_user(db: Session, user_id: int):
    return db.get(User, user_id)

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, user_data: UserCreate):
    db_user = User(name=user_data.name, email=user_data.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)  # DBが採番したidを取得するために必要
    return db_user

def update_user(db: Session, user_id: int, user_data: UserUpdate):
    db_user = db.get(User, user_id)
    if not db_user:
        return None
    update_data = user_data.model_dump(exclude_unset=True)  # 送られたフィールドだけ更新
    for key, value in update_data.items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    db_user = db.get(User, user_id)
    if not db_user:
        return None
    db.delete(db_user)
    db.commit()
    return db_user

main.py ── FastAPIエンドポイント

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import List

import models, schemas, crud
from database import engine, get_db

# テーブルを自動作成（初回起動時）
models.Base.metadata.create_all(bind=engine)

app = FastAPI(title="ユーザー管理API")

# ── 一覧取得 ──────────────────────────────
@app.get("/users", response_model=List[schemas.UserResponse])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

# ── 1件取得 ──────────────────────────────
@app.get("/users/{user_id}", response_model=schemas.UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
    return user

# ── 作成 ─────────────────────────────────
@app.post("/users", response_model=schemas.UserResponse, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    existing = crud.get_user_by_email(db, user.email)
    if existing:
        raise HTTPException(status_code=400, detail="このメールアドレスは既に登録済みです")
    return crud.create_user(db, user)

# ── 更新 ─────────────────────────────────
@app.put("/users/{user_id}", response_model=schemas.UserResponse)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    updated = crud.update_user(db, user_id, user)
    if not updated:
        raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
    return updated

# ── 削除 ─────────────────────────────────
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    deleted = crud.delete_user(db, user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="ユーザーが見つかりません")

💡 Depends(get_db) とは？
FastAPIの Dependency Injection（依存性注入） という仕組みです。
関数が呼ばれるたびに get_db() が自動実行され、DBセッションを受け取れます。
リクエストが終わると自動で db.close() も呼ばれます。

5. SQLModel で書き直す（比較）

SQLModel を使うと、models.py と schemas.py を 1ファイルに統合 できます。

pip install fastapi uvicorn sqlmodel

models.py（SQLModel版）── DBモデルとPydanticモデルを兼ねる

from sqlmodel import SQLModel, Field
from typing import Optional

# テーブル定義 兼 レスポンスモデル
class UserBase(SQLModel):
    name:  str   = Field(min_length=1, max_length=50)
    email: str

class User(UserBase, table=True):          # table=True でDBテーブルになる
    id:        Optional[int] = Field(default=None, primary_key=True)
    is_active: bool = Field(default=True)

class UserCreate(UserBase):                # 作成リクエスト用
    pass

class UserUpdate(SQLModel):               # 更新リクエスト用（全フィールド省略可）
    name:      Optional[str]  = None
    is_active: Optional[bool] = None

class UserResponse(UserBase):             # レスポンス用
    id:        int
    is_active: bool

database.py（SQLModel版）

from sqlmodel import create_engine, Session, SQLModel

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

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

def get_db():
    with Session(engine) as session:
        yield session

SQLAlchemy vs SQLModel コード量比較

ファイル	SQLAlchemy	SQLModel
`models.py`	10行	─
`schemas.py`	20行	─
統合ファイル	─	20行（2ファイル分）
`database.py`	18行	12行

SQLModelは定義ファイルの行数が約半分になります。
一方、SQLAlchemyは複雑なクエリやマイグレーション（Alembic）との連携で優位です。

6. 動かしてみる

uvicorn main:app --reload

http://localhost:8000/docs を開くと、Swagger UIで全エンドポイントを試せます。

動作確認の順番

# 1. ユーザー作成
POST /users
{"name": "Alice", "email": "alice@example.com"}
→ 201 Created: {"id": 1, "name": "Alice", ...}

# 2. 一覧取得
GET /users
→ [{"id": 1, "name": "Alice", ...}]

# 3. 1件取得
GET /users/1
→ {"id": 1, "name": "Alice", ...}

# 4. 更新
PUT /users/1
{"name": "Alice Updated"}
→ {"id": 1, "name": "Alice Updated", ...}

# 5. 削除
DELETE /users/1
→ 204 No Content

# 6. 存在しないIDを取得
GET /users/999
→ 404: {"detail": "ユーザーが見つかりません"}

7. よくあるエラーと対処法

エラー1：`from_attributes` を忘れた

pydantic_core._pydantic_core.ValidationError:
  Value error, PydanticUserError ...

原因： schemas.py の UserResponse に model_config = {"from_attributes": True} がない。
SQLAlchemyオブジェクトをPydanticモデルに変換できていません。

エラー2：`check_same_thread` の設定漏れ

sqlalchemy.exc.ProgrammingError:
  SQLite objects created in a thread can only be used in that same thread.

原因： create_engine() に connect_args={"check_same_thread": False} がない。
SQLiteはデフォルトでスレッド間の共有を禁止しているため、この設定が必要です。

エラー3：`db.refresh()` を忘れた

db_user = User(name=user_data.name, email=user_data.email)
db.add(db_user)
db.commit()
# db.refresh(db_user) を忘れると id が None のまま！
return db_user

原因： commit() 後に refresh() しないと、DBが自動採番した id がPythonオブジェクトに反映されません。

8. まとめ

やったこと	ポイント
SQLAlchemy でCRUD実装	`models.py` と `schemas.py` を分離する構成
SQLModel で書き直し	1クラスでDB定義とPydantic両方を兼ねる
Depends(get_db)	DBセッションの自動管理（DI）
HTTPException	404/400など適切なエラーレスポンス
model_dump(exclude_unset=True)	送られたフィールドだけ部分更新

次回は JWT認証をFastAPIに実装する を解説します。「ログイン済みユーザーだけがアクセスできるエンドポイント」を作ります。

参考

この記事は学習日記として書いています。間違いや補足があればコメントいただけると嬉しいです！