Skip to content

モデル分離

レスポンスにパスワードを含めたりしないように、用途ごとにモデルを分ける。
※コードは認証(bcrypt)からの続き; User クラス1つで全部やろうとしてたせいで本番では使えない感じになってた。

リクエスト用・レスポンス用・テーブル用でモデルを分離する。
SQLModel 公式チュートリアル: https://sqlmodel.tiangolo.com/tutorial/fastapi/multiple-models/


モデルを分ける

models.py を編集して、 User を共通部分の UserBase から派生させつつ、用途ごとのモデルを追加する。

# models.py
from sqlmodel import SQLModel, Field


# 共通のフィールド
class UserBase(SQLModel):
    name: str
    email: str


# データベースのテーブル
class User(UserBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    hashed_password: str


# 作成リクエスト用(平文パスワードを受け取る)
class UserCreate(UserBase):
    password: str


# レスポンス用(hashed_password を含めない)
class UserPublic(UserBase):
    id: int


# 更新リクエスト用(全部任意)
class UserUpdate(SQLModel):
    name: str | None = None
    email: str | None = None
    password: str | None = None

Info

UserBase に共通のフィールド( name , email )をまとめておいて、各モデルはそこに必要なものだけ足す形にしてる。

  • Usertable=True なので、これだけがデータベースのテーブルになる。 hashed_password を持つのはここだけ。
  • UserCreate … ユーザーが送ってくるのは平文の passwordhashed_password は送らせない。
  • UserPublic … レスポンスとして返すモデル。 hashed_password が無いので外に漏れない。
  • UserUpdate … 一部だけ更新できるように、全フィールドを任意( None 許容)にしてる。

Note

テーブルの列( name , email , id , hashed_password )は前のページと変わってないので、 app.db は作り直さなくてOK。


エンドポイントの修正

users.py をモデルに合わせて修正する。

  • 各エンドポイントに response_model=UserPublic を付けて、返り値から hashed_password をなくす
  • create_userUserCreate でパスワードを平文で受け取って、ハッシュ化してから User を作る
# users.py
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from sqlmodel import select

from db import SessionDep
from models import (
    User,
    UserCreate,
    UserPublic,
    UserUpdate,
)
from security import decode_access_token, hash_password

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    session: SessionDep,
) -> User:
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_access_token(token)
        email = payload.get("sub")
        if email is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception

    user = session.exec(select(User).where(User.email == email)).first()
    if user is None:
        raise credentials_exception
    return user


router = APIRouter(
    prefix="/users",
    tags=["users"],
    dependencies=[Depends(get_current_user)],
)


@router.post(
    "/",
    response_model=UserPublic
)
def create_user(
    user: UserCreate,
    session: SessionDep,
):
    hashed_password = hash_password(user.password)
    db_user = User.model_validate(user, update={"hashed_password": hashed_password})
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user


@router.get(
    "/{user_id}",
    response_model=UserPublic
)
def read_user(
    user_id: int,
    session: SessionDep,
):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.patch(
    "/{user_id}",
    response_model=UserPublic,
)
def update_user(
    user_id: int,
    user: UserUpdate,
    session: SessionDep,
):
    db_user = session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    user_data = user.model_dump(exclude_unset=True)
    extra_data = {}
    if "password" in user_data:
        extra_data["hashed_password"] = hash_password(user_data["password"])
    db_user.sqlmodel_update(user_data, update=extra_data)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user


@router.delete("/{user_id}")
def delete_user(
    user_id: int,
    session: SessionDep
):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    session.delete(user)
    session.commit()
    return {"message": "User deleted successfully"}

Info

User.model_validate(user, update={...})UserCreate のデータから User を作りつつ、 hashed_password を足すための書き方。
更新の方は db_user.sqlmodel_update(user_data, update=extra_data) で、送られてきた項目だけ上書きできる( exclude_unset=True で「送られてきた項目だけ」を取り出してる)。


動作確認

curl で、レスポンスから hashed_password が消えてることを確認できるし、 Swagger UI でも確認できる。
まずはJWTを取得してから、ユーザーを取得してみる。

% curl -X 'GET' \
  'http://localhost:8000/users/1' \
  -H 'accept: application/json' \
  -H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'
{"name":"alice","email":"alice@example.com","id":1}

hashed_password が含まれなくなった。
ユーザー作成も、平文の password を送るだけでOKになってる。

% curl -X 'POST' \
  'http://localhost:8000/users/' \
  -H 'accept: application/json' \
  -H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...' \
  -H 'Content-Type: application/json' \
  -d '{"name":"bob","email":"bob@example.com","password":"secret_password"}'
{"name":"bob","email":"bob@example.com","id":2}

Tip

モデルの分け方は Full Stack FastAPI Template のモデル定義 がかなり参考になる。
EmailStr でメールアドレスの形式を検証したり、 Field(max_length=...) で文字数制限を付けたりもできるので、本番ではそのあたりも足していくとよさそう。