📝FastAPI × SQLModelで作るTodoアプリ③：アーキテクチャと実装の詳細

はじめに

どうも、水無月せきなです。

本記事は、シリーズ「FastAPI × SQLModelで作るTodoアプリ」の第３回です。
前回の記事では、DBマイグレーションについて紹介しました。

今回は、アプリの設計思想（クリーンアーキテクチャ）と、それに基づいた具体的な実装について解説していきます。

主な内容は以下の通りです。

• ディレクトリ構成とレイヤーの対応
• FastAPIの依存性注入による依存関係の解決
• モデル、ユースケース、リポジトリ、ルーターの役割と実装

シリーズ一覧

• 🛠️FastAPI × SQLModelで作るTodoアプリ①：開発環境とプロジェクトのセットアップ
• 🗄️FastAPI × SQLModelで作るTodoアプリ②：AlembicによるDBマイグレーション入門
• 📝FastAPI × SQLModelで作るTodoアプリ③：アーキテクチャと実装の詳細
• 🧪FastAPI × SQLModelで作るTodoアプリ④：テストの手法と実装

リポジトリ

開発環境

Windows 11 24H2
WSL2
Docker Desktop 4.43.1
Cursor 1.2.2
Python 3.12
PostgreSQL 15

実装について

ディレクトリ構造

app
├── infrastructure
│   └── db.py
├── __init__.py
├── main.py
├── models
│   ├── __init__.py
│   └── todo.py
├── repositories
│   ├── __init__.py
│   └── todo_repository.py
├── routers
│   ├── __init__.py
│   └── todo.py
├── tests
│   ├── conftest.py
│   ├── infrastructure
│   │   └── test_db.py
│   ├── models
│   │   └── test_todo.py
│   ├── repositories
│   │   └── test_todo_repository.py
│   ├── routers
│   │   └── test_todo_router.py
│   └── usecases
│       └── test_todo_usecase.py
└── usecases
    ├── __init__.py
    └── todo_usecase.py

※マイグレーションなど、一部フォルダ・ファイルは除外しています。

アーキテクチャ

完全に依拠できているわけではないと思うのですが、クリーンアーキテクチャを志向する形で実装しています。

レイヤー	ディレクトリ
Entities	models
Use Cases	Usecases
Interface	routers・repositories
Frameworks	infrastructure

依存関係とその解決

依存関係

依存関係の解決

FastAPIが提供するDIの仕組みを、そのまま使用しています。

依存関係の解決

# app/routers/todo.py
@router.get("/", response_model=list[TodoRead])
def get_todos(todo_usecase: TodoUsecase = Depends()) -> list[TodoRead]:
    return todo_usecase.get_todos()

# app/usecases/todo_usecase.py
class TodoUsecase:
    def __init__(self, todo_repository: TodoRepository = Depends(TodoRepository)) -> None:
        self.todo_repository = todo_repository

# app/repositories/todo_repository.py
class TodoRepository:
    def __init__(self, session: Session = Depends(get_session)) -> None:
        self.session = session

# app/infrastructure/db.py
def get_session() -> Generator[Session, None, None]:
    engine = get_database_engine()
    with Session(engine) as session:
        yield session

依存先のインスタンスをDependsで受け取ることで、連鎖的に依存関係を解決しています。

また、DBのセッションを返すget_sessionでは、withとyieldを使うことで、リクエストの終了時にセッションを開放するようにしています。これは、FastAPIの依存関係解決の仕様を利用した実装です。

実装の詳細

ここからは、具体的な実装について説明したいと思います。

app/models/todo.py（Entities）

データベースやリクエスト・レスポンスで使用するモデルを定義しています。

コード全体

app/models/todo.py

"""
Todoアプリケーションのデータモデル定義

このモジュールはSQLModelを使用してTodoアイテムのデータベースモデルと
API用のPydanticモデルを定義します。
"""

from pydantic import ValidationInfo, field_validator
from sqlmodel import Field, SQLModel


class TodoBase(SQLModel):
    """
    Todoアイテムの基本データ構造

    全てのTodoモデルで共通して使用される、基本的なフィールドを定義します。
    """

    title: str  # Todoアイテムのタイトル
    description: str  # Todoアイテムの詳細説明
    completed: bool = False  # 完了状態（デフォルト: False）


class Todo(TodoBase, table=True):
    """
    データベーステーブル用のTodoモデル

    実際のデータベーステーブルとして使用されるモデルです。
    SQLModelの`table=True`によってテーブルとして認識されます。
    """

    id: int | None = Field(default=None, primary_key=True)  # 主キー（自動採番）


class TodoCreate(TodoBase):
    """
    Todo作成時のリクエストモデル

    新しいTodoアイテムを作成する際のAPIリクエストで使用されます。
    idフィールドは含まれません（自動採番のため）。
    """

    @field_validator("title", "description")
    @classmethod
    def validate_title(cls, v: str, info: ValidationInfo) -> str:
        """タイトルが空文字列でないことを検証する"""
        if not v or v.strip() == "":
            raise ValueError(f"{info.field_name} is required")
        return v


class TodoRead(TodoBase):
    """
    Todo読み取り用のレスポンスモデル

    APIレスポンスでTodoアイテムを返す際に使用されます。
    idフィールドが含まれます。
    """

    id: int  # Todoアイテムの一意識別子


class TodoUpdate(TodoBase):
    """
    Todo更新時のリクエストモデル

    既存のTodoアイテムを更新する際のAPIリクエストで使用されます。
    """

    title: str | None = None  # 更新するタイトル
    description: str | None = None  # 更新する説明
    completed: bool | None = None  # 更新する完了状態

    @field_validator("title", "description")
    @classmethod
    def validate_string_field(cls, v: str | None, info: ValidationInfo) -> str | None:
        """タイトルと説明のバリデーション"""
        # 明示的にNoneが送信された場合はエラー
        if v is None:
            raise ValueError(f"{info.field_name} cannot be null")
        # 空文字列や空白のみの場合もエラー
        if not v or not v.strip():
            raise ValueError(f"{info.field_name} is required")
        return v

定義の仕方としては、FastAPIのチュートリアルを踏襲しています。

異なる点としては、バリデーション用のロジックを追加しています。

app/models/todo.py

class TodoCreate(TodoBase):

    @field_validator("title", "description")
    @classmethod
    def validate_title(cls, v: str, info: ValidationInfo) -> str:
        """タイトルが空文字列でないことを検証する"""
        if not v or v.strip() == "":
            raise ValueError(f"{info.field_name} is required")
        return v


class TodoUpdate(TodoBase):

    title: str | None = None  # 更新するタイトル
    description: str | None = None  # 更新する説明
    completed: bool | None = None  # 更新する完了状態

    @field_validator("title", "description")
    @classmethod
    def validate_string_field(cls, v: str | None, info: ValidationInfo) -> str | None:
        """タイトルと説明のバリデーション"""
        # 明示的にNoneが送信された場合はエラー
        if v is None:
            raise ValueError(f"{info.field_name} cannot be null")
        # 空文字列や空白のみの場合もエラー
        if not v or not v.strip():
            raise ValueError(f"{info.field_name} is required")
        return v

field_validatorを付けた関数は第3引数にValidationInfoを取ることができます。
これにより、バリデーションを行うフィールドに関する情報にアクセスできます。

取得できる情報としては、下記のものがあるようです。

属性	内容
info.field_name	現在検証中のフィールド名（文字列）
info.data	入力データの辞書（型変換前、未検証）
info.field_info	Field(...) で指定した制約（min_length など）

引用元：https://zenn.dev/ikepon/scraps/e0514bfde4386d

値にNoneを設定されると、Pydanticのバリデーションを通り抜けてしまいデータベース保存時にエラーとなったため、カスタムのバリデーションで弾くようにしています。

app/usecases/todo_usecase.py(Use Case）

現時点だと複数のモデルが絡むようなことは無いため、Repositoryに処理を渡して結果をそのまま返しています。
なので、特に触れることはありません。

コード全体

app/usecases/todo_usecase.py

from fastapi import Depends

from app.models.todo import TodoCreate, TodoRead, TodoUpdate
from app.repositories.todo_repository import TodoRepository


class TodoUsecase:app/usecases/todoUsecase.py
    """Todo操作のためのユースケースクラス。

    このクラスは、Todoに関連するビジネスロジックを担当します。
    Clean Architectureのユースケース層として機能し、
    FastAPIの依存性注入システムを通じて提供されます。

    TodoRepositoryを通じてデータベース操作を行い、
    必要に応じてビジネスロジックを適用します。

    Attributes:
        todo_repository (TodoRepository): Todoのデータベース操作を担当するリポジトリ

    Examples:
        FastAPIルーターでの使用例:
        >>> from fastapi import Depends
        >>> from app.usecases.todoUsecase import get_todoUsecase, TodoUsecase
        >>>
        >>> @router.get("/todos")
        >>> def get_all_todos(usecase: TodoUsecase = Depends(get_todoUsecase)):
        ...     return usecase.get_todos()

    Notes:
        - このクラスは依存性注入パターンで使用されます
        - TodoRepositoryを通じてデータベース操作を実行します
        - 各メソッドは適切なエラーハンドリングとバリデーションを含みます
    """

    def __init__(self, todo_repository: TodoRepository = Depends(TodoRepository)) -> None:
        """TodoUsecaseを初期化します。

        Args:
            todo_repository (TodoRepository): Todoのデータベース操作を担当するリポジトリ
        """
        self.todo_repository = todo_repository

    def get_todos(self) -> list[TodoRead]:
        """全てのTodoを取得する。

        データベースから全てのTodoアイテムを取得し、
        表示用のモデルとして返します。

        Returns:
            list[TodoRead]: 全てのTodoを含むリスト。

        Raises:
            データベースアクセスエラーなどの例外が発生する可能性があります。
        """
        return self.todo_repository.get_all_todos()

    def get_todo(self, todo_id: int) -> TodoRead:
        """指定されたIDのTodoを取得する。

        データベースから指定されたIDのTodoアイテムを取得し、
        表示用のモデルとして返します。

        Args:
            todo_id (int): 取得するTodoのID。

        Returns:
            TodoRead: 指定されたTodoアイテム。

        Raises:
            ValueError: 指定されたIDのTodoが見つからない場合
            データベースアクセスエラーなどの例外が発生する可能性があります。
        """
        return self.todo_repository.get_todo(todo_id)

    def create_todo(self, todo_create: TodoCreate) -> TodoRead:
        """新しいTodoを作成する。

        入力された情報を基に新しいTodoアイテムを作成し、
        データベースに保存します。

        Args:
            todo_create (TodoCreate): 作成するTodoの情報。

        Returns:
            TodoRead: 作成されたTodoアイテム。

        Raises:
            データベースアクセスエラーなどの例外が発生する可能性があります。
        """
        return self.todo_repository.create_todo(todo_create)

    def update_todo(self, todo_id: int, todo_update: TodoUpdate) -> TodoRead:
        """指定されたIDのTodoを更新する。

        指定されたIDのTodoアイテムを更新し、
        データベースに変更を保存します。

        Args:
            todo_id (int): 更新するTodoのID。
            todo_update (TodoUpdate): 更新するTodoの内容。

        Returns:
            TodoRead: 更新されたTodoアイテム。

        Raises:
            ValueError: 指定されたIDのTodoが見つからない場合
            データベースアクセスエラーなどの例外が発生する可能性があります。
        """
        return self.todo_repository.update_todo(todo_id, todo_update)

    def delete_todo(self, todo_id: int) -> bool:
        """指定されたIDのTodoを削除する。

        指定されたIDのTodoアイテムをデータベースから削除します。

        Args:
            todo_id (int): 削除するTodoのID。

        Returns:
            bool: 削除が成功した場合True

        Raises:
            ValueError: 指定されたIDのTodoが見つからない場合
            データベースアクセスエラーなどの例外が発生する可能性があります。
        """
        return self.todo_repository.delete_todo(todo_id)

app/repositories/todo_repository.py（Interface）

TodoモデルのCRUD処理を担います。

コード全体

app/repositories/todo_repository.py

from fastapi import Depends
from sqlmodel import Session, select

from app.infrastructure.db import get_session
from app.models.todo import Todo, TodoCreate, TodoRead, TodoUpdate


class TodoRepository:
    """
    Todoアイテムのデータベース操作を担当するリポジトリクラス

    このクラスはTodoアイテムのCRUD操作（作成、読み取り、更新、削除）を提供します。
    """

    def __init__(self, session: Session = Depends(get_session)) -> None:
        """
        TodoRepositoryを初期化します

        Args:
            session: データベースセッション
        """
        self.session = session

    def get_all_todos(self) -> list[TodoRead]:
        """
        すべてのTodoアイテムを取得します

        Returns:
            list[TodoRead]: すべてのTodoアイテムのリスト
        """
        todos = self.session.exec(select(Todo)).all()
        return [TodoRead.model_validate(todo) for todo in todos]

    def create_todo(self, todo: TodoCreate) -> TodoRead:
        """
        新しいTodoアイテムを作成します

        Args:
            todo: 作成するTodoアイテムの情報

        Returns:
            TodoRead: 作成されたTodoアイテム
        """
        # 保存用のモデルを作成
        new_todo = Todo.model_validate(todo)

        # データベースに保存
        self.session.add(new_todo)
        self.session.commit()
        self.session.refresh(new_todo)

        # 保存後のデータで表示用のモデルを返却
        return TodoRead.model_validate(new_todo)

    def _get_todo_by_id(self, todo_id: int) -> Todo:
        """
        指定されたIDのTodoアイテムを取得します

        Args:
            todo_id: 取得するTodoアイテムのID

        Returns:
            Todo: 指定されたTodoアイテム

        Raises:
            ValueError: 指定されたIDのTodoアイテムが見つからない場合
        """
        todo = self.session.get(Todo, todo_id)
        if not todo:
            raise ValueError(f"Todo with id {todo_id} not found")
        return todo

    def get_todo(self, todo_id: int) -> TodoRead:
        """
        指定されたIDのTodoアイテムを取得します

        Args:
            todo_id: 取得するTodoアイテムのID

        Returns:
            TodoRead: 指定されたTodoアイテム

        Raises:
            ValueError: 指定されたIDのTodoアイテムが見つからない場合
        """
        todo = self._get_todo_by_id(todo_id)
        return TodoRead.model_validate(todo)

    def update_todo(self, todo_id: int, todo: TodoUpdate) -> TodoRead:
        """
        指定されたIDのTodoアイテムを更新します

        Args:
            todo_id: 更新するTodoアイテムのID
            todo: 更新する内容

        Returns:
            TodoRead: 更新されたTodoアイテム

        Raises:
            ValueError: 指定されたIDのTodoアイテムが見つからない場合
        """
        # 対象データの取得
        target = self._get_todo_by_id(todo_id)

        # 更新するデータの取得
        update_data = todo.model_dump(exclude_unset=True)

        # 更新するデータを適用
        target.sqlmodel_update(update_data)

        # データベースに保存
        self.session.add(target)
        self.session.commit()
        self.session.refresh(target)

        # 更新後のデータで表示用のモデルを返却
        return TodoRead.model_validate(target)

    def delete_todo(self, todo_id: int) -> bool:
        """
        指定されたIDのTodoアイテムを削除します

        Args:
            todo_id: 削除するTodoアイテムのID

        Returns:
            bool: 削除が成功した場合True

        Raises:
            ValueError: 指定されたIDのTodoアイテムが見つからない場合
        """
        # 対象データの取得
        todo = self._get_todo_by_id(todo_id)

        # データベースから削除
        self.session.delete(todo)
        self.session.commit()

        return True

基本的にはFastAPIのチュートリアルを踏襲しています。

異なる点としては、

• 内部的にTodoを取得する共通関数の定義（_get_todo_by_id）
• TodoRead.model_validateによる読み取り専用モデルの返却

が挙げられると思います。

app/repositories/todo_repository.py

class TodoRepository:

    def _get_todo_by_id(self, todo_id: int) -> Todo:
        todo = self.session.get(Todo, todo_id)
        if not todo:
            raise ValueError(f"Todo with id {todo_id} not found")
        return todo

    def get_todo(self, todo_id: int) -> TodoRead:
        todo = self._get_todo_by_id(todo_id)
        return TodoRead.model_validate(todo)

    def delete_todo(self, todo_id: int) -> bool:
        # 対象データの取得
        todo = self._get_todo_by_id(todo_id)

        # データベースから削除
        self.session.delete(todo)
        self.session.commit()

        return True

app/infrastructure/db.py（Frameworks）

DBへの接続を扱う部分です。

コード全体

app/infrastructure/db.py

import os
from collections.abc import Generator
from functools import lru_cache

from sqlalchemy import Engine
from sqlmodel import Session, create_engine


@lru_cache
def get_database_engine() -> Engine:
    """データベースエンジンを取得する（キャッシュあり）。

    環境変数からデータベース接続情報を取得し、SQLAlchemyエンジンを作成します。
    @lru_cacheデコレータにより、一度作成されたエンジンは再利用されます。

    Returns:
        Engine: SQLAlchemyのエンジンインスタンス

    Raises:
        KeyError: 必要な環境変数が設定されていない場合
        ValueError: データベース接続文字列の構築に失敗した場合
    """
    try:
        postgres_user = os.environ["POSTGRES_USER"]
        postgres_password = os.environ["POSTGRES_PASSWORD"]
        postgres_server = os.environ["POSTGRES_SERVER"]
        postgres_port = os.environ["POSTGRES_PORT"]
        postgres_db = os.environ["POSTGRES_DB"]
    except KeyError as e:
        raise KeyError(f"Required environment variable not set: {e}") from e

    if not all([postgres_user, postgres_password, postgres_server, postgres_port, postgres_db]):
        raise ValueError("One or more database environment variables are empty")

    database_url = f"postgresql://{postgres_user}:{postgres_password}@{postgres_server}:{postgres_port}/{postgres_db}"
    return create_engine(database_url)


def get_session() -> Generator[Session, None, None]:
    """データベースセッションを提供する。

    FastAPIの依存性注入システムで使用されるジェネレータ関数です。
    with文とyieldを組み合わせることで、セッションの自動クローズを保証します。

    Yields:
        Session: SQLModelのセッションインスタンス

    Examples:
        FastAPIでの使用例:
        >>> from fastapi import Depends
        >>> def get_todos(session: Session = Depends(get_session)):
        ...     return session.exec(select(Todo)).all()

    Notes:
        - セッションは自動的にクローズされるため、手動でclose()を呼ぶ必要はありません
        - エラーが発生した場合も、with文によりセッションは適切にクローズされます
    """
    engine = get_database_engine()
    with Session(engine) as session:
        yield session

依存関係の解決で触れましたが、withとyieldを使うことで、リクエストの終了時にセッションが開放されるようにしています。

また、@lru_cacheでキャッシュを返すようにしていますが、これはClaudeの提案です。
シングルトンよりこちらの方が良いとか何とか……

app/routers/todo.py（Interface）

ルーティングを行う部分です。
リクエストを受け取り、ユースケースを呼び出します。

details

app/routers/todo.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.exc import IntegrityError

from app.models.todo import TodoCreate, TodoRead, TodoUpdate
from app.usecases.todoUsecase import TodoUsecase

router = APIRouter(prefix="/todos", tags=["todos"])


# ruff: noqa
@router.get("/", response_model=list[TodoRead])
def get_todos(todo_usecase: TodoUsecase = Depends()) -> list[TodoRead]:
    return todo_usecase.get_todos()


@router.get("/{todo_id}", response_model=TodoRead)
def get_todo(todo_id: int, todo_usecase: TodoUsecase = Depends()) -> TodoRead:
    try:
        return todo_usecase.get_todo(todo_id)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))


@router.post("/", response_model=TodoRead)
def create_todo(todo_create: TodoCreate, todo_usecase: TodoUsecase = Depends()) -> TodoRead:
    try:
        return todo_usecase.create_todo(todo_create)
    except IntegrityError as e:
        raise HTTPException(
            status_code=400, detail="Failed to create todo due to data constraint violation"
        )


@router.put("/{todo_id}", response_model=TodoRead)
def update_todo(
    todo_id: int, todo_update: TodoUpdate, todo_usecase: TodoUsecase = Depends()
) -> TodoRead:
    try:
        return todo_usecase.update_todo(todo_id, todo_update)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except IntegrityError as e:
        raise HTTPException(
            status_code=400, detail="Failed to update todo due to data constraint violation"
        )


@router.delete("/{todo_id}", response_model=bool)
def delete_todo(todo_id: int, todo_usecase: TodoUsecase = Depends()) -> bool:
    try:
        return todo_usecase.delete_todo(todo_id)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

基本的に呼び出しているだけですが、エラーが返ってきそうな場所はハンドリングしてHTTPExceptionを返すようにしています。

おわりに

ここまでで、クリーンアーキテクチャを意識した構成と、各レイヤーの具体的な実装について紹介しました。
次回は、pytestやその他のライブラリを用いたテストコードの実装方法について解説します。

ぜひ次回も合わせてお読みください！

参考資料

• Pydanticを調べて使ってみる