0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@qq2016(LJ Q)

【FastAPI入門 #4】FastAPI + SQLiteでCRUDアプリを作る ── SQLAlchemy & SQLModel 両方で書く

0
Posted at

はじめに

前回までで、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に実装する を解説します。「ログイン済みユーザーだけがアクセスできるエンドポイント」を作ります。

参考

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

0
0
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
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?