ChatGPT先生と挑むFastAPI #10｜総合課題：タスク管理APIを自力実装

これまで学んできたFastAPIの知識を使い、総合課題としてタスク管理APIを実装します。

APIやデータベースの設計からCRUD処理、テストまで、できるだけ自分で考えながら実装していきます。

さらにDocker Composeを使ってFastAPIとPostgreSQLを起動し、これまでの学習内容を一通り組み合わせて確認します。

このSTEPでは、FastAPIを使ったAPIを自分で設計・実装できる状態になることを目指します。

ChatGPT先生

それではSTEP10を始めましょう！

総合課題の要件
作業項目
完了条件・最終レビュー観点
STEP 10の最終ゴール
課題開始
1. 作業順
レビュー結果
まとめ
成果物

■■■スポンサーリンク■■■

総合課題の要件

ChatGPT先生

以下をSTEP 10の課題仕様として、これを基準に進めます。

作成するもの

タスク管理API

既存のSTEP 1〜9で使用してきたFastAPIプロジェクトをそのまま使用してよい。
既存のproductsなどの実装も削除する必要はない。

使用する技術

これまでの学習で使用してきた以下の技術を利用する。新しいライブラリを追加する必要はない。

FastAPI
Pydantic
SQLAlchemy
PostgreSQL
Alembic
pytest
Docker / Docker Compose

タスクが持つデータ

タスクには最低限、以下の情報を持たせる。

項目	要件

ID	タスクを一意に識別できる。DB側で自動採番
タイトル	必須。最大文字数100文字
説明	任意。最大文字数1000文字
完了状態	`true / false`で管理。新規作成時は未完了

これ以外の項目を追加する必要はない。

必要なAPI

以下の操作ができるAPIを実装する。

API	要件
タスク登録	新しいタスクを登録できること
タスク一覧取得	登録されているタスクを一覧取得できること `limit`と`offset`を指定して取得範囲を変更できること
タスク1件取得	IDを指定してタスクを1件取得できること
タスク更新	IDを指定して既存タスクを更新できること
タスク削除	IDを指定して既存タスクを削除できること

エラー処理

最低限、以下に対応する。

存在しないタスクIDを指定した場合、404 Not Found
リクエスト内容がPydanticの定義を満たさない場合、適切なバリデーションエラー

データベース

PostgreSQLを使用する。

SQLAlchemyでタスクを扱えるようにし、必要なテーブルはAlembicによるマイグレーションで作成する。

APIテスト

pytestを使って、自分で必要だと考えたAPIテストを実装する。

単に正常系だけを通すのではなく、「このAPIが要件どおり動作していることを保証するには何をテストすべきか」を自分で考えてテストケースを決める。

Docker

最終的にDocker Compose環境で、

FastAPI
PostgreSQL

を起動し、アプリケーションが正常に利用できる状態にする。

今回の対象外

以下は実装しない。

認証・認可
ユーザー管理
フロントエンド
タスク期限
カテゴリ・タグ
担当者
PATCH API
ページネーション用の高度なレスポンス形式
新しい外部ライブラリの導入

また、設計書などの特別な成果物は作成しない。

実装前の設計は、要件を整理して自分で実装方針を考えるためのものとする。

作業項目

ChatGPT先生

以下の内容で課題に取り組んでください。

1. 課題要件を確認する

この仕様を確認し、実装対象を把握する。

2. 要件を整理し、タスク管理APIを自力で実装する

要件から、以下のような内容を自分で考えて実装する。

必要なAPI
DB構造
Model
Schema
Router
ファイル構成
エラー処理
実装順序

ChatGPT先生から実装順序は指定しない。

Swagger UIなどを使った途中の動作確認は、必要に応じて自由に行ってよい。

3. 必要なpytestを自分で考えて実装する

完成したAPIについて、何をテストすれば要件どおり動作していると判断できるかを自分で考え、テストを実装する。

4. pytestを実行し、不具合を修正する

テストを実行する。

失敗したテストがある場合は、自分で原因を調査して実装またはテストを修正する。

最終的に必要なテストがすべて成功する状態にする。

5. Docker Compose環境で最終確認する

Docker ComposeでFastAPIとPostgreSQLを起動し、コンテナ環境でも正常に動作することを確認する。

6. 完成した実装をレビューする

実装・pytest・Docker環境まで完成した段階で、ChatGPT先生が全体をレビューする。

原則として、このレビューまではChatGPT先生側から完成形を提示しない。

レビュー時に提出するもの

レビューでは、以下を確認できる状態で提出する。

提出物	内容
FastAPIの実装コード	タスク管理APIに関係するRouter、Schema、Model、DB処理など
`main.py`などアプリ起動に関係するコード	Routerの登録やFastAPIアプリ全体の構成を確認できるもの
Alembic関連ファイル	タスク用テーブルを作成したマイグレーションファイルと、必要に応じて設定ファイル
pytestのテストコード	タスク管理APIに対して作成したテスト一式
pytest実行結果	すべてのテストが成功していることを確認できる実行結果
Docker関連ファイル	`Dockerfile`、`compose.yaml`など実行環境を確認できるファイル
環境設定に関係するファイル	`.env.example`や設定クラスなど、DB接続やアプリ設定の構成を確認できるもの
Swagger UI / OpenAPIの確認結果	APIドキュメントが正しく生成されていることを確認できるスクリーンショットなど
Docker Composeの動作確認結果	FastAPIとPostgreSQLが正常に起動していることを確認できる実行結果

.envなど、パスワードや秘密情報を含むファイルそのものは提出しない。

設計書などの追加資料は不要とし、実装されたコードと動作確認結果をレビュー対象とする。

完了条件・最終レビュー観点

ChatGPT先生

以下をすべて確認できたらSTEP 10完了とします。

要件

タスク管理APIに必要な機能がすべて実装されている
課題で指定されていない機能を不必要に追加していない

API

タスクの登録ができる
タスク一覧を取得できる
limit / offsetによる取得範囲の変更ができる
IDによる1件取得ができる
タスクを更新できる
タスクを削除できる
存在しないIDに対して適切な404が返る
不正な入力に対して適切なバリデーションエラーが返る

FastAPI / Pydantic

Schemaの使い分けが適切
型定義やバリデーションが要件に合っている
FastAPIの機能を不自然な方法で回避していない
APIの入出力仕様が適切に定義されている

OpenAPI / Swagger UI

実装したAPIがOpenAPIドキュメントに正しく生成されている
HTTPメソッドとパスが意図した仕様になっている
リクエストボディの仕様が正しく表示されている
レスポンスの仕様が適切に反映されている
パスパラメータ・クエリパラメータが正しく認識されている

SQLAlchemy / PostgreSQL

DBモデルが要件に合っている
Sessionの扱いが適切
CRUD処理が正しくDBへ反映される
不要・不自然なDBアクセスがない

Alembic

タスク用テーブルをマイグレーションで作成できる
ModelとDBスキーマに矛盾がない

エラー処理

存在しないデータの扱いが適切
HTTPステータスコードが不自然ではない
必要な場所で適切にエラーを返している

pytest

テストがすべて成功する
正常系だけでなく、必要な異常系も確認されている
API要件を保証するために必要なテストケースが考えられている
テスト同士が不必要に依存していない
テストDB・fixtureなど、STEP 8で学んだ仕組みが適切に使用されている

Docker

Docker ComposeでFastAPIとPostgreSQLを起動できる
FastAPIからPostgreSQLへ接続できる
コンテナを作り直しても必要な構成を再現できる
Docker環境固有の設定に不自然な点がない

コード品質

Router・Model・Schemaなどの責務が不自然に混在していない
不要な処理や重複コードがない
コードから処理の意図を追える
これまで学んだ構成や仕組みを正しく使えている
学習用としてだけ動く特殊な実装ではなく、実務につながる自然な構成になっている

STEP 10の最終ゴール

ChatGPT先生

完成コードを作ることだけがゴールではありません。

要件を受け取ったときに、自分で必要な実装を考え、APIを作り、テストによって動作を保証し、Docker環境まで完成させられること。

ここまで自力でできる状態になっているかを確認します。

課題開始

作業順

次のような順番で実装しました。

実際の作業順序

タスクテーブルの定義を作成

models/task.pyの作成

実装したコード↓

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True, comment="ID")
    title: Mapped[str] = mapped_column(String(100), comment="タイトル")
    explanation: Mapped[str | None] = mapped_column(String(1000), nullable=True, comment="説明")
    status: Mapped[bool] = mapped_column(default=False, comment="完了状態")

APIのリクエストボディの定義を作成

shemas/tasks.pyの作成

実装したコード↓

from pydantic import BaseModel, Field


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    explanation: str | None = Field(default=None, max_length=1000)
    status: bool = False


class TaskUpdate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    explanation: str | None = Field(default=None, max_length=1000)
    status: bool = False

3
タスクテーブルのマイグレーション
マイグレーションファイルの生成
```
alembic revision --autogenerate -m "create task table"
```
DBにタスクテーブルを反映
```
alembic upgrade head
```

APIの作成

routers/tasks.pyの作成
※「5. テストコードの作成」と並行して、動作確認をしながら実装。

タスク登録：POST /tasks
タスク一覧取得：GET /tasks
タスク取得：GET /tasks/{task_id}
タスク更新：PUT /tasks/{task_id}
タスク削除：DELETE /tasks/{task_id}

実装したコード↓

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.dependencies import get_db, get_pagination
from app.models.task import Task
from app.schemas.tasks import TaskCreate, TaskUpdate

router = APIRouter()


@router.post(
    "/tasks",
    status_code=status.HTTP_201_CREATED,
)
def create_task(
    task_data: TaskCreate,
    db: Annotated[Session, Depends(get_db)],
):
    # リクエストボディから登録用のモデルを作成
    new_task = Task(
        title=task_data.title,
        explanation=task_data.explanation,
        status=task_data.status,
    )
    # Sessionに登録用モデルを追加
    db.add(new_task)
    # 登録を確定
    db.commit()
    # 登録した内容を取得
    db.refresh(new_task)

    return {
        "id": new_task.id,
        "title": new_task.title,
        "explanation": new_task.explanation,
        "status": new_task.status,
    }


@router.get("/tasks")
def get_tasks(
    pagination: Annotated[dict, Depends(get_pagination)],
    db: Annotated[Session, Depends(get_db)],
):
    # ページング条件を取得
    limit = pagination["limit"]
    offset = pagination["offset"]

    stmt = (
        select(Task)
        .order_by(Task.id)
        .offset(offset)
        .limit(limit)
    )
    result = db.execute(stmt)

    rows = result.scalars().all()

    # 取得結果を辞書に変換
    tasks = []
    for row in rows:
        task = {
            "id": row.id,
            "title": row.title,
            "explanation": row.explanation,
            "status": row.status,
        }

        tasks.append(task)

    return tasks


@router.get(
    "/tasks/{task_id}",
    responses={
        404: {
            "description": "Task not found",
        }
    }
)
def get_task(
    task_id: int,
    db: Annotated[Session, Depends(get_db)],
):
    # 主キーを指定してテーブルからタスクを取得
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(
            status_code=404,
            detail="Task not found",
        )

    return {
        "id": task.id,
        "title": task.title,
        "explanation": task.explanation,
        "status": task.status,
    }


@router.put(
    "/tasks/{task_id}",
    responses={
        404: {
            "description": "Task not found",
        }
    }
)
def update_task(
    task_id: int,
    task_data: TaskUpdate,
    db: Annotated[Session, Depends(get_db)],
):
    # 更新対象のタスクをテーブルから取得
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(
            status_code=404,
            detail="Task not found",
        )

    # 更新データの作成
    task.title = task_data.title
    task.explanation = task_data.explanation
    task.status = task_data.status

    # コミット
    db.commit()
    db.refresh(task)

    return {
        "id": task.id,
        "title": task.title,
        "explanation": task.explanation,
        "status": task.status,
    }


@router.delete(
    "/tasks/{task_id}",
    responses={
        404: {
            "description": "Task not found",
        }
    }
)
def delete_task(
    task_id: int,
    db: Annotated[Session, Depends(get_db)],
):
    # 削除対象のタスクをテーブルから取得
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(
            status_code=404,
            detail="Task not found",
        )

    # タスクを削除対象とする
    db.delete(task)
    # コミット
    db.commit()

    return {
        "message": "Task deleted",
    }

main.pyの修正（修正箇所のみ抜粋）

from app.routers.tasks import router as tasks_router

・・・省略・・・

# タスクAPIのRouterをFastAPIアプリへ登録する
app.include_router(tasks_router)

テストコードの作成

tests/test_tasks/pyの作成
※「4. APIの作成」と並行して実装。

＜POST /tasksのテストケース＞

正常系：タスクの登録
正常系：リクエストデータのexplanationとstatusを省略してタスクの登録
異常系：リクエストデータのtitleをブランクでタスクの登録
異常系：リクエストデータのtitleをNoneでタスクの登録
異常系：リクエストデータのtitleが最大文字数オーバーでタスクの登録
異常系：リクエストデータのexplanationが最大文字数オーバーでタスクの登録

＜GET /tasksのテストケース＞

正常系：タスク一覧の取得
正常系：データが存在しない場合
正常系：limitがゼロの場合

＜GET /tasks/{task_id}のテストケース＞

正常系：指定したタスクの取得
異常系：指定したタスクが存在しない場合
異常系：パスパラメータが不正の場合

＜PUT /tasks/{task_id}のテストケース＞

正常系：タスクの更新
正常系：リクエストデータのexplanationとstatusを省略してタスクの更新
異常系：指定したタスクが存在しない場合
異常系：パスパラメータが不正の場合
異常系：リクエストデータのtitleをブランクでタスクの更新
異常系：リクエストデータのtitleをNoneでタスクの更新
異常系：リクエストデータのtitleが最大文字数オーバーでタスクの更新
異常系：リクエストデータのexplanationが最大文字数オーバーでタスクの更新

＜DELETE /tasks/{task_id}のテストケース＞

正常系：タスクの削除
異常系：指定したタスクが存在しない場合
異常系：パスパラメータが不正の場合

実装したテストコード↓

from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.task import Task


def test_create_task(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：タスクの登録"""
    # リクエストデータを作成
    title = "a" * 100
    assert len(title) == 100
    explanation = "a" * 1000
    assert len(explanation) == 1000
    request_data = {
        "title": title,
        "explanation": explanation,
        "status": True,
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 201

    # レスポンスボディを検証
    response_data = response.json()
    task_id = response_data["id"]

    assert response_data == {
        "id": task_id,
        "title": title,
        "explanation": explanation,
        "status": True,
    }

    # DBに登録されていることを検証
    task = db_session.get(Task, task_id)
    assert task is not None
    assert task.title == title
    assert task.explanation == explanation
    assert task.status == True


def test_create_task_defaults_explanation_and_status_when_omitted(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：リクエストデータのexplanationとstatusを省略してタスクの登録"""
    # リクエストデータを作成
    request_data = {
        "title": "テストタスク",
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 201

    # レスポンスボディを検証
    response_data = response.json()
    task_id = response_data["id"]

    assert response_data == {
        "id": task_id,
        "title": "テストタスク",
        "explanation": None,
        "status": False,
    }

    # DBに登録されていることを検証
    task = db_session.get(Task, task_id)
    assert task is not None
    assert task.title == "テストタスク"
    assert task.explanation == None
    assert task.status == False


def test_create_task_blank_title(
        client: TestClient,
) -> None:
    """異常系：リクエストデータのtitleをブランクでタスクの登録"""
    # リクエストデータを作成
    request_data = {
        "title": "",
        "explanation": "テストタスクの説明",
        "status": True,
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # エラーが発生した項目を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "title"]
        for error in errors
    )


def test_create_task_none_title(
        client: TestClient,
) -> None:
    """異常系：リクエストデータのtitleをNoneでタスクの登録"""
    # リクエストデータを作成
    request_data = {
        "title": None,
        "explanation": "テストタスクの説明",
        "status": True,
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # エラーが発生した項目を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "title"]
        for error in errors
    )


def test_create_task_max_len_over_title(
        client: TestClient,
) -> None:
    """異常系：リクエストデータのtitleが最大文字数オーバーでタスクの登録"""
    # リクエストデータを作成
    data = "a" * 101
    assert len(data) > 100

    request_data = {
        "title": data,
        "explanation": "テストタスクの説明",
        "status": True,
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # エラーが発生した項目を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "title"]
        for error in errors
    )


def test_create_task_max_len_over_explanation(
        client: TestClient,
) -> None:
    """異常系：リクエストデータのexplanationが最大文字数オーバーでタスクの登録"""
    # リクエストデータを作成
    data = "a" * 1001
    assert len(data) > 1000

    request_data = {
        "title": "テストタスク",
        "explanation": data,
        "status": True,
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # エラーが発生した項目を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "explanation"]
        for error in errors
    )


def test_get_task_list(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：タスク一覧の取得"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(10):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i <= 5
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # APIを実行してタスク一覧を取得（5件目から3件取得）
    response = client.get(
        "/tasks",
        params={
            "limit": 3,
            "offset": 4,
        }
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを取得
    res_tasks = response.json()

    # 件数の検証
    assert len(res_tasks) == 3

    # レスポンスボディを検証
    assert res_tasks == [
        {
            "id": tasks[4].id,
            "title": "タスク05",
            "explanation": "タスク05の説明",
            "status": True,
        },
        {
            "id": tasks[5].id,
            "title": "タスク06",
            "explanation": "タスク06の説明",
            "status": True,
        },
        {
            "id": tasks[6].id,
            "title": "タスク07",
            "explanation": "タスク07の説明",
            "status": False,
        },
    ]


def test_get_task_list_not_found(
        client: TestClient,
) -> None:
    """正常系：データが存在しない場合"""
    # APIを実行してタスク一覧を取得
    response = client.get(
        "/tasks",
        params={
            "limit": 1,
            "offset": 0,
        }
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    assert response.json() == []


def test_get_task_list_limit_zero(
        client: TestClient,
) -> None:
    """正常系：limitがゼロの場合"""
    # APIを実行してタスク一覧を取得
    response = client.get(
        "/tasks",
        params={
            "limit": 0,
            "offset": 0,
        }
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    assert response.json() == []


def test_get_task(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：指定したタスクの取得"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # APIを実行してタスクを取得
    response = client.get(f"/tasks/{tasks[1].id}")

    # HTTPステータスコードを検証する
    assert response.status_code == 200

    # レスポンスの内容を検証する
    assert response.json() == {
        "id": tasks[1].id,
        "title": "タスク02",
        "explanation": "タスク02の説明",
        "status": False,
    }


def test_get_task_not_found(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：指定したタスクが存在しない場合"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # APIを実行してタスクを取得
    response = client.get("/tasks/999")

    # HTTPステータスコードを検証する
    assert response.status_code == 404

    # レスポンスの内容を検証する
    assert response.json() == {
        "detail": "Task not found",
    }


def test_get_task_invalid_id(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：パスパラメータが不正の場合"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # APIを実行してタスクを取得
    response = client.get("/tasks/abc")

    # HTTPステータスコードを検証する
    assert response.status_code == 422

    # レスポンスの内容を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["path", "task_id"]
        for error in errors
    )


def test_update_task(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：タスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": "更新タスク",
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[2].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    response_data = response.json()

    assert response_data == {
        "id": tasks[2].id,
        "title": "更新タスク",
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # DBに変更が反映されていることを検証
    task = db_session.get(Task, tasks[2].id)
    assert task is not None
    assert task.title == "更新タスク"
    assert task.explanation == "更新タスクの説明"
    assert task.status == True


def test_update_task_defaults_explanation_status_when_omitted(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：リクエストデータのexplanationとstatusを省略してタスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": "更新タスク",
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[0].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    response_data = response.json()

    assert response_data == {
        "id": tasks[0].id,
        "title": "更新タスク",
        "explanation": None,
        "status": False,
    }

    # DBに変更が反映されていることを検証
    task = db_session.get(Task, tasks[0].id)
    assert task is not None
    assert task.title == "更新タスク"
    assert task.explanation == None
    assert task.status == False


def test_update_task_not_found(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：指定したタスクが存在しない場合"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": "更新タスク",
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        "/tasks/999",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 404

    # レスポンスの内容を検証する
    assert response.json() == {
        "detail": "Task not found",
    }


def test_update_task_invalid_id(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：パスパラメータが不正の場合"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": "更新タスク",
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        "/tasks/abc",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # レスポンスの内容を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["path", "task_id"]
        for error in errors
    )


def test_update_task_blank_title(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：リクエストデータのtitleをブランクでタスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": "",
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[0].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # レスポンスの内容を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "title"]
        for error in errors
    )


def test_update_task_none_title(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：リクエストデータのtitleをブランクでタスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": None,
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[0].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # レスポンスの内容を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "title"]
        for error in errors
    )


def test_update_task_max_len_over_title(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：リクエストデータのtitleが最大文字数オーバーでタスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    data = "a" * 101
    assert len(data) > 100

    request_data = {
        "title": data,
        "explanation": "更新タスクの説明",
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[0].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # レスポンスの内容を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "title"]
        for error in errors
    )


def test_update_task_max_len_over_explanation(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：リクエストデータのexplanationが最大文字数オーバーでタスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    data = "a" * 1001
    assert len(data) > 1000

    request_data = {
        "title": "更新タスク",
        "explanation": data,
        "status": True,
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[0].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # レスポンスの内容を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "explanation"]
        for error in errors
    )


def test_delete_task(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：タスクの削除"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # DELETE APIを実行
    response = client.delete(f"/tasks/{tasks[2].id}")

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    assert  response.json() == {
        "message": "Task deleted",
    }

    # DBから削除対象のタスクを検索する
    deleted_task = db_session.execute(
        select(Task).where(Task.id == tasks[2].id)
    ).scalar_one_or_none()

    # タスクがDBに存在しないことを検証する
    assert deleted_task is None


def test_delete_task_not_found(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：指定したタスクが存在しない場合"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # DELETE APIを実行
    response = client.delete("/tasks/999")

    # HTTPステータスコードを検証
    assert response.status_code == 404

    # レスポンスボディを検証
    assert  response.json() == {
        "detail": "Task not found",
    }


def test_delete_task_invalid_id(
        client: TestClient,
        db_session: Session,
) -> None:
    """異常系：パスパラメータが不正の場合"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # DELETE APIを実行
    response = client.delete("/tasks/abc")

    # HTTPステータスコードを検証
    assert response.status_code == 422

    # レスポンスボディを検証
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["path", "task_id"]
        for error in errors
    )

テストの実行結果↓

6
Swagger UIの確認

実装したAPIがSwagger UIに表示されているか確認

＜POST /tasks＞

＜GET /tasks＞

＜GET /tasks/{task_id}＞

＜PUT /tasks/{task_id}＞

＜DELETE /tasks/{task_id}＞

アプリ実行用環境でpytestを実行

Docker compseをビルド

docker compose -f compose.run.yaml build app

実行用環境のコンテナを起動

docker compose -f compose.run.yaml up -d

pytestを実行

> docker compose -f compose.run.yaml exec app python -m pytest -v
======================================================== test session starts ========================================================
platform linux -- Python 3.12.14, pytest-9.1.1, pluggy-1.6.0 -- /usr/local/bin/python
cachedir: .pytest_cache
rootdir: /app
plugins: asyncio-1.4.0, anyio-4.14.2
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 37 items

tests/test_database.py::test_connect_to_test_database PASSED                                                                  [  2%]
tests/test_database.py::test_products_table_exists PASSED                                                                     [  5%]
tests/test_products.py::test_product_create_rollback PASSED                                                                   [  8%]
tests/test_products.py::test_get_product PASSED                                                                               [ 10%]
tests/test_products.py::test_create_product PASSED                                                                            [ 13%]
tests/test_products.py::test_update_product PASSED                                                                            [ 16%]
tests/test_products.py::test_delete_product PASSED                                                                            [ 18%]
tests/test_products.py::test_get_product_not_found PASSED                                                                     [ 21%]
tests/test_products.py::test_get_product_invalid_id PASSED                                                                    [ 24%]
tests/test_products.py::test_update_product_not_found PASSED                                                                  [ 27%]
tests/test_products.py::test_delete_product_not_found PASSED                                                                  [ 29%]
tests/test_products.py::test_create_product_invalid_price PASSED                                                              [ 32%]
tests/test_products.py::test_create_product_missing_name PASSED                                                               [ 35%]
tests/test_products.py::test_duplicate_product PASSED                                                                         [ 37%]
tests/test_tasks.py::test_create_task PASSED                                                                                  [ 40%]
tests/test_tasks.py::test_create_task_defaults_explanation_and_status_when_omitted PASSED                                     [ 43%]
tests/test_tasks.py::test_create_task_blank_title PASSED                                                                      [ 45%]
tests/test_tasks.py::test_create_task_none_title PASSED                                                                       [ 48%]
tests/test_tasks.py::test_create_task_max_len_over_title PASSED                                                               [ 51%]
tests/test_tasks.py::test_create_task_max_len_over_explanation PASSED                                                         [ 54%]
tests/test_tasks.py::test_get_task_list PASSED                                                                                [ 56%]
tests/test_tasks.py::test_get_task_list_not_found PASSED                                                                      [ 59%]
tests/test_tasks.py::test_get_task_list_limit_zero PASSED                                                                     [ 62%]
tests/test_tasks.py::test_get_task PASSED                                                                                     [ 64%]
tests/test_tasks.py::test_get_task_not_found PASSED                                                                           [ 67%]
tests/test_tasks.py::test_get_task_invalid_id PASSED                                                                          [ 70%]
tests/test_tasks.py::test_update_task PASSED                                                                                  [ 72%]
tests/test_tasks.py::test_update_task_defaults_explanation_status_when_omitted PASSED                                         [ 75%]
tests/test_tasks.py::test_update_task_not_found PASSED                                                                        [ 78%]
tests/test_tasks.py::test_update_task_invalid_id PASSED                                                                       [ 81%]
tests/test_tasks.py::test_update_task_blank_title PASSED                                                                      [ 83%]
tests/test_tasks.py::test_update_task_none_title PASSED                                                                       [ 86%]
tests/test_tasks.py::test_update_task_max_len_over_title PASSED                                                               [ 89%]
tests/test_tasks.py::test_update_task_max_len_over_explanation PASSED                                                         [ 91%]
tests/test_tasks.py::test_delete_task PASSED                                                                                  [ 94%]
tests/test_tasks.py::test_delete_task_not_found PASSED                                                                        [ 97%]
tests/test_tasks.py::test_delete_task_invalid_id PASSED                                                                       [100%]

======================================================== 37 passed in 0.70s =========================================================

8
レビュー依頼

ソースコードやテスト結果、Swagger UIのキャプチャなどをChatGPT先生に提出。
9
レビュー指摘の修正

下記の「レビュー結果」に記載の指摘を対応。
10
再レビュー依頼 → 完了

修正したリソースを再提出。

問題ないと判定され課題完了。

レビュー結果

※以下のレビュー指摘の修正は、すべて上記のコードに反映済みです。

指摘事項1：explanationのDB文字数

ChatGPT先生

ModelはString(100)なのに、Schemaは最大1000文字です。APIでは101〜1000文字を許可するのにDBでは保存できない、という不整合があります

指摘の修正

titleをコピペしてexplanationを書いた時に桁数を修正していなかったことが原因。

models/task.pyのexplanationを以下のように修正。

explanation: Mapped[str] = mapped_column(String(1000), nullable=True, comment="説明")

マイグレーションの実行

alembic revision --autogenerate -m "edit explanation length task table"

エラーが発生

ERROR [alembic.util.messaging] Target database is not up to date.
  FAILED: Target database is not up to date.

一度マイグレーションを最新状態にしてから再実行したら成功。

alembic upgrade head

alembic revision --autogenerate -m "edit explanation length task table"

変更をDBに反映

alembic upgrade head

POST /tasksのテストコードを以下に修正して、最大桁数で登録できることを確認

def test_create_task(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：タスクの登録"""
    # リクエストデータを作成
    title = "a" * 100
    assert len(title) == 100
    explanation = "a" * 1000
    assert len(explanation) == 1000
    request_data = {
        "title": title,
        "explanation": explanation,
        "status": True,
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 201

    # レスポンスボディを検証
    response_data = response.json()
    task_id = response_data["id"]

    assert response_data == {
        "id": task_id,
        "title": title,
        "explanation": explanation,
        "status": True,
    }

    # DBに登録されていることを検証
    task = db_session.get(Task, task_id)
    assert task is not None
    assert task.title == title
    assert task.explanation == explanation
    assert task.status == True

実行結果

$ python -m pytest tests/test_tasks.py::test_create_task -v
========================================================================================================= test session starts ==========================================================================================================
platform linux -- Python 3.12.14, pytest-9.1.1, pluggy-1.6.0 -- /usr/local/bin/python
cachedir: .pytest_cache
rootdir: /workspace
plugins: asyncio-1.4.0, anyio-4.14.2
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 1 item                                                                                                                                                                                                                       

tests/test_tasks.py::test_create_task PASSED                                                                                                                                                                                     [100%]

========================================================================================================== 1 passed in 0.13s ===========================================================================================================

指摘事項2：explanationが実質必須

ChatGPT先生

explanation: str | None = Field(max_length=1000)は「nullを許可する必須項目」です。今回の要件は説明が任意なので、省略可能にする必要があります

指摘の修正

純粋に勘違いしていました。

shemas/tasks.pyを以下のように修正

from pydantic import BaseModel, Field


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    explanation: str | None = Field(default=None, max_length=1000)
    status: bool = False


class TaskUpdate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    explanation: str | None = Field(default=None, max_length=1000)
    status: bool = False

PUTリクエストのときにexplanationを省略した場合は、Noneで更新すべきか、更新しないべきか確認したところ、今回は「Noneで更新する」という回答が返ってきたため、そのように対応します。（と言ってもAPI側は修正不要）

テストコードの修正（該当のケースのみ抜粋）

def test_create_task_defaults_explanation_and_status_when_omitted(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：リクエストデータのexplanationとstatusを省略してタスクの登録"""
    # リクエストデータを作成
    request_data = {
        "title": "テストタスク",
    }

    # POST APIを実行
    response = client.post(
        "/tasks",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 201

    # レスポンスボディを検証
    response_data = response.json()
    task_id = response_data["id"]

    assert response_data == {
        "id": task_id,
        "title": "テストタスク",
        "explanation": None,
        "status": False,
    }

    # DBに登録されていることを検証
    task = db_session.get(Task, task_id)
    assert task is not None
    assert task.title == "テストタスク"
    assert task.explanation == None
    assert task.status == False


def test_update_task_defaults_explanation_status_when_omitted(
        client: TestClient,
        db_session: Session,
) -> None:
    """正常系：リクエストデータのexplanationとstatusを省略してタスクの更新"""
    # テスト用のデータを作成する
    tasks = []
    for i in range(3):
        task_no = str(i + 1).zfill(2)
        task = Task(
            title=f"タスク{task_no}",
            explanation=f"タスク{task_no}の説明",
            status=i % 2 == 0
        )
        tasks.append(task)

    # テストデータを一括登録
    db_session.add_all(tasks)
    db_session.commit()

    # リクエストデータを作成
    request_data = {
        "title": "更新タスク",
    }

    # PUT APIを実行
    response = client.put(
        f"/tasks/{tasks[0].id}",
        json=request_data,
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    response_data = response.json()

    assert response_data == {
        "id": tasks[0].id,
        "title": "更新タスク",
        "explanation": None,
        "status": False,
    }

    # DBに変更が反映されていることを検証
    task = db_session.get(Task, tasks[0].id)
    assert task is not None
    assert task.title == "更新タスク"
    assert task.explanation == None
    assert task.status == False

指摘事項3：タスク一覧取得でlimit=0 の場合に [] が返ること

テストケースが漏れているため追加します。

テストケース↓

def test_get_task_list_limit_zero(
        client: TestClient,
) -> None:
    """正常系：limitがゼロの場合"""
    # APIを実行してタスク一覧を取得
    response = client.get(
        "/tasks",
        params={
            "limit": 0,
            "offset": 0,
        }
    )

    # HTTPステータスコードを検証
    assert response.status_code == 200

    # レスポンスボディを検証
    assert response.json() == []

テスト結果↓

$ python -m pytest tests/test_tasks.py::test_get_task_list_limit_zero -v
========================================================================================================= test session starts ==========================================================================================================
platform linux -- Python 3.12.14, pytest-9.1.1, pluggy-1.6.0 -- /usr/local/bin/python
cachedir: .pytest_cache
rootdir: /workspace
plugins: asyncio-1.4.0, anyio-4.14.2
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 1 item                                                                                                                                                                                                                       

tests/test_tasks.py::test_get_task_list_limit_zero PASSED                                                                                                                                                                        [100%]

========================================================================================================== 1 passed in 0.17s ===========================================================================================================