基礎から学ぶPython入門 90日コース | データベース・Web API - Day 70:CRUD API

Python 90日で身につけるPython
スポンサーリンク
スポンサーリンク

Day 70:CRUD API ― 「APIでデータを育てて、読み書きして、片付ける」

Day 70では、いよいよ CRUD API をひとまとまりの形で作っていきます。

キーワードはこの4つです。

  • Create(作る)
  • Read(読む)
  • Update(更新する)
  • Delete(削除する)

ここまでで、

  • データベース(SQLite・SQL)
  • ORM(モデルとCRUD)
  • FastAPI(GET / POST、リクエストボディ、バリデーション)

と、必要なピースはほぼ揃いました。

今日はそれらを組み合わせて、 「ひとつのリソース(例えば顧客やタスク)を、APIでCRUDできるようにする」 という、Web/API開発の王道パターンを体験していきます。

CRUD APIの全体像をイメージする

何をCRUDするAPIにするか

題材はシンプルに、「タスク管理」 にしましょう。

  • タスクを作る(Create)
  • タスク一覧を読む(Read)
  • タスクを更新する(Update)
  • タスクを削除する(Delete)

という、よくある「ToDo管理API」を作ります。

今回は、データベースではなく、 メモリ上のリスト を使って実装します。

理由はシンプルで、

  • CRUDの流れに集中できる
  • コードが短くて読みやすい
  • 「DBに変えるときのイメージ」も持ちやすい

からです。

データモデルを決める ― 「タスクって、どんな情報を持っている?」

タスクの項目を決める

タスク管理なので、最低限こんな項目が欲しくなります。

  • id:タスクの識別子(整数)
  • title:タイトル(必須)
  • description:詳細(任意)
  • done:完了フラグ(真偽値)

これを、FastAPI+Pydanticで表現していきます。

基本セットアップ ― FastAPIとPydanticモデル

# day70_crud_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class TaskBase(BaseModel):
    """
    タスクの基本情報を表すモデルです。
    クライアントから送られてくるデータの「共通部分」を定義します。
    """

    title: str
    description: str | None = None
    done: bool = False


class TaskCreate(TaskBase):
    """
    タスク作成用のモデルです。
    TaskBase と同じですが、id は含みません(サーバー側で採番するため)。
    """
    pass


class Task(TaskBase):
    """
    レスポンス用のタスクモデルです。
    サーバー側で管理する id を含みます。
    """

    id: int
Python

ここでのポイントは、

  • TaskBase:共通の項目(title / description / done)
  • TaskCreate:作成時に使うモデル(idなし)
  • Task:レスポンスや内部管理に使うモデル(idあり)

という役割分担です。

この「Base+Create+Response」の分け方は、 実際のAPI設計でもよく使われるパターンです。

データの置き場所 ― メモリ上の「なんちゃってDB」

シンプルな「タスク一覧」を用意する

# メモリ上にタスクを保存するためのリストです。
# 実際のアプリでは、ここがデータベースに置き換わります。
tasks: list[Task] = []

# id を採番するためのカウンタです。
next_id: int = 1
Python
  • tasks:タスクの一覧(Taskインスタンスのリスト)
  • next_id:新しいタスクに割り当てるIDを管理するカウンタ

この2つが、今回の「なんちゃってDB」です。

Create ― タスクを作るAPI

POST /tasks で新しいタスクを作る

@app.post("/tasks", response_model=Task)
def create_task(task_create: TaskCreate):
    """
    新しいタスクを作成するためのPOST APIです。

    - リクエストボディとして TaskCreate を受け取ります。
    - サーバー側で id を採番し、Task として保存します。
    - 作成されたタスクをレスポンスとして返します。
    """

    global next_id

    # 新しい Task インスタンスを作成します。
    task = Task(
        id=next_id,
        title=task_create.title,
        description=task_create.description,
        done=task_create.done,
    )

    # メモリ上のリストに追加します。
    tasks.append(task)

    # 次のタスクのために id をインクリメントしておきます。
    next_id += 1

    return task
Python

ここでの流れは、

  1. クライアントから TaskCreate 形式のJSONが送られてくる
  2. FastAPI+Pydanticがバリデーションして task_create に変換
  3. サーバー側で id を採番して Task インスタンスを作る
  4. tasks リストに追加する
  5. 作成されたタスクをレスポンスとして返す

という、Createの王道パターン です。

Read ― タスクを読むAPI

GET /tasks で一覧を取得する

@app.get("/tasks", response_model=list[Task])
def list_tasks():
    """
    タスク一覧を取得するためのGET APIです。

    - メモリ上の tasks リストをそのまま返します。
    - response_model に list[Task] を指定することで、
      レスポンスが Task のリストとして整形されます。
    """

    return tasks
Python
  • response_model=list[Task] によって、 「Taskのリストとして返す」という意図が明確になります。

GET /tasks/{task_id} で1件取得する

@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int):
    """
    指定したIDのタスクを取得するためのGET APIです。

    - パスパラメータ task_id を受け取ります。
    - tasks リストから該当するタスクを探します。
    - 見つからなければ 404 を返します。
    """

    for task in tasks:
        if task.id == task_id:
            return task

    # 見つからなかった場合は HTTPException を投げて 404 を返します。
    raise HTTPException(status_code=404, detail="Task not found")
Python

ここでのポイントは、

  • パスパラメータ task_id: int を受け取る
  • tasks リストをループして該当IDを探す
  • 見つからなければ HTTPException で 404 を返す

という、Read(1件取得)の基本パターン です。

Update ― タスクを更新するAPI

PUT /tasks/{task_id} でタスクを上書きする

更新にはいくつかのスタイルがありますが、 ここでは「指定したタスクを丸ごと更新する」PUT を使ってみます。

@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, task_update: TaskCreate):
    """
    指定したIDのタスクを更新するためのPUT APIです。

    - パスパラメータ task_id で対象を指定します。
    - リクエストボディ task_update で新しい内容を受け取ります。
    - 該当タスクが見つかれば、内容を上書きします。
    - 見つからなければ 404 を返します。
    """

    for index, task in enumerate(tasks):
        if task.id == task_id:
            # 新しい Task インスタンスを作成して、既存の位置に上書きします。
            updated_task = Task(
                id=task_id,
                title=task_update.title,
                description=task_update.description,
                done=task_update.done,
            )
            tasks[index] = updated_task
            return updated_task

    raise HTTPException(status_code=404, detail="Task not found")
Python

ここでの流れは、

  1. task_id で対象タスクを特定
  2. task_update: TaskCreate で新しい内容を受け取る
  3. 該当タスクを見つけたら、新しい Task で上書き
  4. 見つからなければ 404

という、Updateの基本パターン です。

Delete ― タスクを削除するAPI

DELETE /tasks/{task_id} でタスクを消す

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    """
    指定したIDのタスクを削除するためのDELETE APIです。

    - パスパラメータ task_id で対象を指定します。
    - 該当タスクが見つかれば、tasks リストから削除します。
    - 見つからなければ 404 を返します。
    - 成功時には簡単なメッセージを返します。
    """

    for index, task in enumerate(tasks):
        if task.id == task_id:
            # 該当タスクをリストから削除します。
            tasks.pop(index)
            return {"message": f"Task {task_id} deleted"}

    raise HTTPException(status_code=404, detail="Task not found")
Python

DELETEでは、

  • 対象を特定して
  • リストから削除し
  • 成功メッセージを返す

という、シンプルだけれど重要な流れを押さえます。

Day 70ミニテンプレート ― CRUD APIひとまとめ

ここまでのコードを、ひとつのファイルにまとめたテンプレートです。

# day70_crud_api_template.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class TaskBase(BaseModel):
    title: str
    description: str | None = None
    done: bool = False


class TaskCreate(TaskBase):
    pass


class Task(TaskBase):
    id: int


tasks: list[Task] = []
next_id: int = 1


@app.get("/")
def read_root():
    return {"message": "Day 70: CRUD API 入門へようこそ"}


@app.post("/tasks", response_model=Task)
def create_task(task_create: TaskCreate):
    global next_id

    task = Task(
        id=next_id,
        title=task_create.title,
        description=task_create.description,
        done=task_create.done,
    )

    tasks.append(task)
    next_id += 1

    return task


@app.get("/tasks", response_model=list[Task])
def list_tasks():
    return tasks


@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int):
    for task in tasks:
        if task.id == task_id:
            return task

    raise HTTPException(status_code=404, detail="Task not found")


@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, task_update: TaskCreate):
    for index, task in enumerate(tasks):
        if task.id == task_id:
            updated_task = Task(
                id=task_id,
                title=task_update.title,
                description=task_update.description,
                done=task_update.done,
            )
            tasks[index] = updated_task
            return updated_task

    raise HTTPException(status_code=404, detail="Task not found")


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    for index, task in enumerate(tasks):
        if task.id == task_id:
            tasks.pop(index)
            return {"message": f"Task {task_id} deleted"}

    raise HTTPException(status_code=404, detail="Task not found"}
Python

このファイルを保存して、

bash

uvicorn day70_crud_api_template:app --reload

と実行すれば、

  • POST /tasks → Create
  • GET /tasks → Read(一覧)
  • GET /tasks/{id} → Read(1件)
  • PUT /tasks/{id} → Update
  • DELETE /tasks/{id} → Delete

という、CRUD APIの基本セット を一通り試せます。

Day 70のまとめ ― 「APIが、データの“ライフサイクル”を支える」

今日の主役は、

  • Create:新しいデータを作る
  • Read:既存のデータを読む
  • Update:データを更新する
  • Delete:不要になったデータを削除する

という、CRUDの4つの動きでした。

FastAPI+Pydanticを使うことで、

  • リクエストボディをモデルとして受け取り
  • バリデーション済みのデータとして扱い
  • レスポンスもモデルに基づいて整形し
  • エラー時にはHTTPステータスとメッセージを返す

という、APIとしての基本的な振る舞い を、 比較的少ないコードで実現できることが分かったと思います。

今日作ったタスク管理CRUD APIは、 そのままデータベースやORMに差し替えることで、 「本当に永続化されるタスク管理サービス」に育てていくことができます。

CRUDは、どんなアプリでもほぼ必ず登場する、 静かだけれど、ものすごく重要な“基礎体力”です。

Day 70でその感覚をつかんだことで、 この先のWeb/API開発が、ぐっと現実味を帯びて見えてくるはずです。

タイトルとURLをコピーしました