基礎から学ぶPython入門 90日コース | データベース・Web API － Day 75：APIプロジェクト

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

Day 75：APIプロジェクト ― 「自分の手でToDo APIを組み上げる」日

Day 75では、ここまで学んできた FastAPI × SQLite × CRUD ×API設計 を ひとつの「小さなプロジェクト」として形にしていきます。

テーマはとても身近なものです。

ToDo API

機能の流れはこうでしたね。

  • ユーザー ↓
  • タスク作成 ↓
  • タスク取得 ↓
  • タスク更新 ↓
  • タスク削除 ↓
  • SQLite保存

今日は、この流れを 一本のAPIとしてちゃんと動く形 にしていきます。

全体設計 ― 「どんなToDo APIにするか」を決める

扱うデータを整理する

まずは、ToDoタスクの情報を整理します。

タスクに必要そうな項目を挙げてみると、こんな感じでしょうか。

  • id：タスクのID（主キー、自動採番）
  • title：タスクのタイトル（必須）
  • description：詳細説明（任意）
  • done：完了フラグ（True / False）
  • created_at：作成日時（任意、今回は省略してもOK）

今回は、初心者向けに少しシンプルにして、

  • id
  • title
  • description（任意）
  • done（デフォルトは False）

という構成で進めます。

エンドポイント設計

次に、APIの「住所」を決めます。

ToDoタスクというリソースに対して、 基本的なCRUDを提供するので、こんな感じが自然です。

  • 一覧取得：GET /tasks
  • 1件取得：GET /tasks/{task_id}
  • 作成：POST /tasks
  • 更新：PUT /tasks/{task_id}
  • 削除：DELETE /tasks/{task_id}

Day 72で学んだ「リソース＋HTTPメソッド」の組み合わせが、 ここでそのまま活きてきます。

SQLiteとSQLAlchemyでタスクテーブルを作る

必要なライブラリ

まずは、使うライブラリを確認しておきます。

  • fastapi
  • uvicorn
  • sqlalchemy
  • pydantic（FastAPIに含まれている形で使われます）

インストール例：

bash

pip install fastapi "uvicorn[standard]" sqlalchemy

データベース接続とベースクラス

# day75_todo_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import declarative_base, sessionmaker, Session

# FastAPIアプリケーションのインスタンスを作成します。
app = FastAPI()

# SQLAlchemyのベースクラス（全モデルの親）を定義します。
Base = declarative_base()

# SQLiteのデータベースファイルへの接続URLを指定します。
DATABASE_URL = "sqlite:///day75_todo.db"

# エンジン（DBとの接続オブジェクト）を作成します。
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLite＋マルチスレッド対策
)

# セッション工場を作成します。これを使ってDBとの会話窓口を作ります。
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Python

ここまでで、

  • SQLiteファイル day75_todo.db に接続する準備
  • モデルの親クラス Base
  • セッションを作るための SessionLocal

が整いました。

タスクテーブルのモデル定義

class Task(Base):
    """
    tasks テーブルに対応するSQLAlchemyモデルです。
    タスクのID、タイトル、説明、完了フラグを持ちます。
    """

    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    title = Column(String(200), nullable=False)       # タイトルは必須
    description = Column(String(500), nullable=True)  # 説明は任意
    done = Column(Boolean, nullable=False, default=False)  # 完了フラグ（デフォルトFalse）
Python
  • __tablename__ = "tasks"：テーブル名
  • id：主キー＋自動採番
  • title：NOT NULL
  • description：NULL許可
  • done：必須（デフォルトはFalse）

テーブル作成

# モデル定義に基づいて、SQLite上にテーブルを作成します。
Base.metadata.create_all(bind=engine)
Python

これで、day75_todo.db に tasks テーブルが作られます。

PydanticモデルでAPIの入出力を整える

リクエスト・レスポンスの形を決める

APIの入出力を表現するために、 Pydanticモデルを定義します。

class TaskBase(BaseModel):
    """
    タスク情報の共通部分を表すPydanticモデルです。
    APIの入出力で使う「形」を定義します。
    """

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


class TaskCreate(TaskBase):
    """
    タスク作成時のリクエストボディ用モデルです。
    id はサーバー側で採番するため含めません。
    """
    pass


class TaskRead(TaskBase):
    """
    タスク情報をレスポンスとして返すためのモデルです。
    DB側で管理している id を含みます。
    """

    id: int

    class Config:
        orm_mode = True
        # orm_mode=True にすることで、
        # SQLAlchemyモデルから直接このPydanticモデルに変換できるようになります。
Python

ここでのポイントは、

  • TaskBase：共通項目（title / description / done）
  • TaskCreate：作成時の入力用
  • TaskRead：レスポンス用（id付き）
  • orm_mode = True：SQLAlchemyモデルをそのままPydanticに変換できるようにする設定

です。

セッション取得のヘルパー（シンプル版）

def get_db() -> Session:
    """
    SQLAlchemyのセッション（DBとの会話窓口）を取得するためのヘルパー関数です。
    FastAPIのエンドポイント内で呼び出して使います。
    """

    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
Python

本格的には Depends(get_db) を使う形がよく登場しますが、 Day 75では、初心者向けに 明示的に SessionLocal() を呼ぶスタイル で進めていきます。

タスク作成 ― POST /tasks

「ユーザー → タスク作成」の入り口

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

    - リクエストボディとして TaskCreate を受け取ります。
    - SQLAlchemyの Task モデルに変換してDBに保存します。
    - 保存されたタスク情報を TaskRead として返します。
    """

    db: Session = SessionLocal()

    # タイトルが空文字の場合は、異常系として扱い、400を返します。
    if len(task_in.title.strip()) == 0:
        db.close()
        raise HTTPException(status_code=400, detail="Title must not be empty")

    # Task モデルのインスタンスを作成します。
    task = Task(
        title=task_in.title,
        description=task_in.description,
        done=task_in.done,
    )

    # セッションに追加してコミットします。
    db.add(task)
    db.commit()
    db.refresh(task)  # DB側で確定した値（idなど）を反映します。

    db.close()

    # Pydanticモデル（TaskRead）として返します。
    return task
Python

ここでの流れは、

  1. TaskCreate を受け取る
  2. タイトルが空でないかチェック（簡単なバリデーション）
  3. Task モデルを作成
  4. db.add() → db.commit() で保存
  5. db.refresh() で id を反映
  6. task をそのまま返す（orm_mode=True により自動変換）

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

タスク取得 ― GET /tasks と GET /tasks/{task_id}

一覧取得 ― GET /tasks

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

    - DBから全タスクを取得します。
    - TaskRead のリストとして返します。
    """

    db: Session = SessionLocal()
    tasks = db.query(Task).order_by(Task.id.asc()).all()
    db.close()

    return tasks
Python
  • 全タスクを id 昇順で取得
  • リストとして返す

1件取得 ― GET /tasks/{task_id}

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

    - パスパラメータ task_id を受け取ります。
    - DBから該当タスクを検索します。
    - 見つからなければ 404 を返します。
    """

    db: Session = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()
    db.close()

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

    return task
Python
  • filter(Task.id == task_id).first() で1件取得
  • 見つからない場合は HTTPException で404

タスク更新 ― PUT /tasks/{task_id}

「タスク更新」の流れを丁寧に

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

    - パスパラメータ task_id で対象を指定します。
    - リクエストボディ task_in で新しい内容を受け取ります。
    - 該当タスクが存在しなければ 404 を返します。
    - タイトルが空の場合は 400 を返します。
    """

    db: Session = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()

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

    if len(task_in.title.strip()) == 0:
        db.close()
        raise HTTPException(status_code=400, detail="Title must not be empty")

    # フィールドを更新します。
    task.title = task_in.title
    task.description = task_in.description
    task.done = task_in.done

    db.commit()
    db.refresh(task)
    db.close()

    return task
Python

Updateでは、

  • 対象タスクの存在チェック
  • タイトルのバリデーション
  • フィールドの上書き
  • commit()＋refresh() で反映

という、慎重な更新の流れ を体験します。

タスク削除 ― DELETE /tasks/{task_id}

「タスク削除」の最後の一歩

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

    - パスパラメータ task_id で対象を指定します。
    - 該当タスクが存在しなければ 404 を返します。
    - 削除後、簡単なメッセージを返します。
    """

    db: Session = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()

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

    db.delete(task)
    db.commit()
    db.close()

    return {"message": f"Task {task_id} deleted"}
Python

Deleteでは、

  • 対象タスクの存在チェック
  • db.delete() → db.commit() で削除
  • 成功メッセージを返す

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

Day 75ミニテンプレート ― ToDo APIひとまとめ

ここまでのコードを、ひとつのファイルにまとめたテンプレートです。 これをそのまま保存して動かせば、ToDo APIプロジェクト が完成します。

# day75_todo_api_template.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import declarative_base, sessionmaker, Session

app = FastAPI()

Base = declarative_base()
DATABASE_URL = "sqlite:///day75_todo.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)


class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    description = Column(String(500), nullable=True)
    done = Column(Boolean, nullable=False, default=False)


Base.metadata.create_all(bind=engine)


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


class TaskCreate(TaskBase):
    pass


class TaskRead(TaskBase):
    id: int

    class Config:
        orm_mode = True


@app.get("/")
def read_root():
    """
    ToDo APIのトップ。簡単なメッセージを返します。
    """
    return {"message": "Day 75: ToDo API プロジェクトへようこそ"}


@app.post("/tasks", response_model=TaskRead)
def create_task(task_in: TaskCreate):
    db: Session = SessionLocal()

    if len(task_in.title.strip()) == 0:
        db.close()
        raise HTTPException(status_code=400, detail="Title must not be empty")

    task = Task(
        title=task_in.title,
        description=task_in.description,
        done=task_in.done,
    )

    db.add(task)
    db.commit()
    db.refresh(task)
    db.close()

    return task


@app.get("/tasks", response_model=list[TaskRead])
def list_tasks():
    db: Session = SessionLocal()
    tasks = db.query(Task).order_by(Task.id.asc()).all()
    db.close()
    return tasks


@app.get("/tasks/{task_id}", response_model=TaskRead)
def get_task(task_id: int):
    db: Session = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()
    db.close()

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

    return task


@app.put("/tasks/{task_id}", response_model=TaskRead)
def update_task(task_id: int, task_in: TaskCreate):
    db: Session = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()

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

    if len(task_in.title.strip()) == 0:
        db.close()
        raise HTTPException(status_code=400, detail="Title must not be empty")

    task.title = task_in.title
    task.description = task_in.description
    task.done = task_in.done

    db.commit()
    db.refresh(task)
    db.close()

    return task


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    db: Session = SessionLocal()
    task = db.query(Task).filter(Task.id == task_id).first()

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

    db.delete(task)
    db.commit()
    db.close()

    return {"message": f"Task {task_id} deleted"}
Python

実行はいつものように、

uvicorn day75_todo_api_template:app --reload

とすればOKです。

  • POST /tasks：タスク作成
  • GET /tasks：タスク一覧
  • GET /tasks/{id}：タスク1件取得
  • PUT /tasks/{id}：タスク更新
  • DELETE /tasks/{id}：タスク削除

という、ToDo APIの一連の流れ を、 自分の手で叩いて確かめることができます。

Day 75のまとめ ― 「小さなAPIでも、“ちゃんと動くサービス”の顔をしている」

今日の主役は、

  • ToDo API という身近な題材
  • タスク作成・取得・更新・削除・SQLite保存 という一連の流れ
  • FastAPI × SQLite × SQLAlchemy × Pydantic の組み合わせ

でした。

ここまでの学びをぎゅっと詰め込んだこの小さなプロジェクトは、 規模こそ控えめですが、 「ちゃんと動くWebサービスの芯」 を、すでに持っています。

  • エンドポイント設計
  • CRUDの流れ
  • バリデーション
  • エラーハンドリング
  • データベース永続化

それぞれが、Day 70〜Day 74で触れてきた内容と、 きれいにつながっているはずです。

Day 75でこのToDo APIを作りきったことで、 「APIを学んでいる人」から、 「自分でAPIプロジェクトを組み立てられる人」 へと、 一歩、確かに進んでいます。

スポンサーリンク
タイトルとURLをコピーしました