• HOME
  • PythonでREST APIを構築する方法（FastAPI）｜CRUD・入力検証・APIキー認証・テストまで

2026年10月10日

  • シェア
  • ツイート

    • 2026年10月10日

    PythonでREST APIを構築する方法（FastAPI）｜CRUD・入力検証・APIキー認証・テストまで

    FastAPI を使うと、Python の型ヒントを書くだけで入力の検証と API ドキュメントの生成まで行われるため、REST API を少ないコードで組み立てられます。この記事では、タスク管理の小さな API を題材に、一覧・取得・作成・更新・削除（CRUD）のエンドポイント、Pydantic による入力検証、SQLite への保存、API キーによる書き込み制限、自動テスト、本番で動かすときの構成までを順に説明します。

    準備：インストールと最初のエンドポイント

    プロジェクト用の仮想環境を作り、FastAPI を入れます。fastapi[standard] を指定すると、開発用サーバーを起動する fastapi コマンドや、テストに使う HTTP クライアントなどもまとめて入ります。

    python3 -m venv .venv
source .venv/bin/activate
pip install "fastapi[standard]"

    この記事のコードは FastAPI の 0.1xx 系と Pydantic 2 系、Python 3.10 以降を前提にしています（2026年10月現在）。FastAPI はまだ 1.0 に達しておらず、細かな推奨の書き方が更新されることがあるので、新しく始めるときは公式ドキュメントのチュートリアルも合わせて確認してください。

    まずは動作確認用の最小構成です。

    # main.py
from fastapi import FastAPI

app = FastAPI()


@app.get("/health")
def health():
    return {"status": "ok"}

    fastapi dev main.py で起動し、ブラウザで http://127.0.0.1:8000/health を開くと JSON が返ります。http://127.0.0.1:8000/docs を開くと、エンドポイントの一覧と試し打ちができる画面（Swagger UI）が自動で作られているのが分かります。fastapi dev はファイルを保存すると自動で再起動する開発用のモードで、本番では使いません。

    データの形を Pydantic で決める

    REST API で最初に決めるのは「どんな JSON を受け取り、どんな JSON を返すか」です。FastAPI では Pydantic のモデルとして書きます。

    from pydantic import BaseModel, Field


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    done: bool = False


class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    done: bool | None = None


class Task(BaseModel):
    id: int
    title: str
    done: bool

    作成用（TaskCreate）、部分更新用（TaskUpdate）、返却用（Task）を分けているのがポイントです。作成時に id を受け取ってしまう、更新時に全項目を送らないとエラーになる、といった問題を避けられます。

    モデルに合わない JSON が送られてくると、FastAPI は処理を呼ぶ前に 422 エラーと、どの項目がなぜ不正かを返します。空文字のタイトルや、文字数の上限を超える値をはじく処理を自分で書く必要はありません。

    データベース接続を依存関係として渡す

    保存先には、Python に標準で入っている SQLite を使います。接続の作成と終了は「依存関係（Depends）」として定義し、各エンドポイントで受け取ります。

    import os
import sqlite3
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI

DB_PATH = os.getenv("DB_PATH", "tasks.db")


@asynccontextmanager
async def lifespan(app: FastAPI):
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS tasks ("
        "id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, done INTEGER NOT NULL DEFAULT 0)"
    )
    conn.commit()
    conn.close()
    yield


app = FastAPI(title="Task API", version="1.0.0", lifespan=lifespan)


def get_db():
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()


DB = Annotated[sqlite3.Connection, Depends(get_db)]

    lifespan は起動時と終了時の処理を書く場所で、ここではテーブルの作成だけを行っています。get_db() は yield の前で接続を作り、レスポンスを返した後に finally で閉じます。こうしておくと、途中で例外が起きても接続が残り続けません。

    check_same_thread=False を付けているのは、FastAPI が async を付けない通常の関数をスレッドプールで実行し、依存関係とエンドポイントが別のスレッドで動くことがあるためです。接続はリクエストごとに作って閉じているので、複数のリクエストで同じ接続を共有することはありません。

    CRUD のエンドポイントを作る

    HTTP メソッドと URL の組み合わせで操作を表すのが REST の基本です。

    • GET /tasks：一覧
    • GET /tasks/{id}：1件取得
    • POST /tasks：作成（成功時は 201）
    • PATCH /tasks/{id}：一部の項目を更新
    • DELETE /tasks/{id}：削除（成功時は 204、本文なし）
    from fastapi import HTTPException, Query, status


def row_to_task(row: sqlite3.Row) -> Task:
    return Task(id=row["id"], title=row["title"], done=bool(row["done"]))


@app.get("/tasks", response_model=list[Task])
def list_tasks(
    db: DB,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
):
    rows = db.execute(
        "SELECT id, title, done FROM tasks ORDER BY id LIMIT ? OFFSET ?", (limit, offset)
    ).fetchall()
    return [row_to_task(r) for r in rows]


@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int, db: DB):
    row = db.execute("SELECT id, title, done FROM tasks WHERE id = ?", (task_id,)).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return row_to_task(row)

    一覧の limit には上限（100）を付けています。上限がないと、1回のリクエストで全件を返すことになり、データが増えたときにサーバーの負荷が大きくなります。

    SQL の値はすべて ? のプレースホルダーで渡しています。f 文字列で SQL に値を埋め込むと SQL インジェクションの原因になるため、型ヒントで int に絞っている場合でもこの書き方を守ります。

    書き込みを API キーで制限する

    作成・更新・削除を誰でも呼べる状態は危険です。ここでは、リクエストヘッダー X-API-Key に正しいキーがあるときだけ書き込みを許可します。

    import secrets
from fastapi import Security
from fastapi.security import APIKeyHeader

API_KEY = os.getenv("API_KEY", "")
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


def require_api_key(key: Annotated[str | None, Security(api_key_header)]):
    if not API_KEY or not key or not secrets.compare_digest(key, API_KEY):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")


@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED,
          dependencies=[Depends(require_api_key)])
def create_task(task: TaskCreate, db: DB):
    cur = db.execute("INSERT INTO tasks (title, done) VALUES (?, ?)", (task.title, int(task.done)))
    db.commit()
    return Task(id=cur.lastrowid, title=task.title, done=task.done)


@app.patch("/tasks/{task_id}", response_model=Task, dependencies=[Depends(require_api_key)])
def update_task(task_id: int, patch: TaskUpdate, db: DB):
    current = get_task(task_id, db)
    data = current.model_dump() | patch.model_dump(exclude_unset=True)
    db.execute("UPDATE tasks SET title = ?, done = ? WHERE id = ?",
               (data["title"], int(data["done"]), task_id))
    db.commit()
    return Task(**data)


@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT,
            dependencies=[Depends(require_api_key)])
def delete_task(task_id: int, db: DB):
    cur = db.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
    db.commit()
    if cur.rowcount == 0:
        raise HTTPException(status_code=404, detail="Task not found")

    設計上の注意点です。

    • キーはコードに書かない：環境変数 API_KEY から読みます。未設定のときは書き込みをすべて拒否するようにしているので、設定漏れで誰でも書き込める状態にはなりません。
    • **比較は secrets.compare_digest()**：== での比較より、文字列の比較にかかる時間からキーを推測される余地を減らせます。
    • **部分更新は exclude_unset=True**：送られてきた項目だけを上書きします。これがないと、送っていない項目が None で上書きされます。

    利用者ごとに権限を分けたい、ログインと組み合わせたいといった場合は、OAuth2 や JWT を使う方法が公式ドキュメントのセキュリティの章で説明されています。認証の仕組みを独自に作り込むより、そうした定番の方式に寄せるほうが安全です。

    テストを書いて動作を確かめる

    FastAPI にはサーバーを起動せずに API を呼べる TestClient があります。pytest と組み合わせると、変更のたびに一通りの動きを確かめられます。

    # test_main.py
import os
import tempfile

os.environ["DB_PATH"] = os.path.join(tempfile.mkdtemp(), "test.db")
os.environ["API_KEY"] = "test-key"

from fastapi.testclient import TestClient
from main import app

H = {"X-API-Key": "test-key"}


def test_task_flow():
    with TestClient(app) as client:
        assert client.post("/tasks", json={"title": "x"}).status_code == 401
        r = client.post("/tasks", json={"title": "記事を書く"}, headers=H)
        assert r.status_code == 201
        task_id = r.json()["id"]
        assert client.post("/tasks", json={"title": ""}, headers=H).status_code == 422
        r = client.patch(f"/tasks/{task_id}", json={"done": True}, headers=H)
        assert r.json()["done"] is True
        assert client.delete(f"/tasks/{task_id}", headers=H).status_code == 204
        assert client.get(f"/tasks/{task_id}").status_code == 404

    pip install pytest のあと pytest で実行します。with TestClient(app) の形で使うと lifespan が実行され、テーブルが作られた状態でテストが始まります。テスト用のデータベースは一時フォルダーに作り、本番のデータに触れないようにしています。

    本番で動かすときの構成

    公開するときは、開発用の fastapi dev ではなく fastapi run か Uvicorn で起動し、その前に Nginx などのリバースプロキシを置く構成が一般的です。

    fastapi run main.py --host 127.0.0.1 --port 8000 --workers 2
    • アプリは 127.0.0.1 で待ち受ける：外部からの接続は Nginx が受け、HTTPS もそこで終端します。
    • CORS は必要なオリジンだけ許可する：ブラウザから別ドメインで呼ぶ場合は fastapi.middleware.cors の CORSMiddleware で allow_origins に許可するドメインを列挙します。"*" で全許可するのは公開データだけを返す API に限ります。
    • ドキュメント画面を閉じる：社内向けの API などで /docs を公開したくない場合は、FastAPI(docs_url=None, redoc_url=None) で無効にできます。
    • SQLite の限界：書き込みが多い、複数台で動かすといった場合は PostgreSQL や MySQL に移します。移行を見据えるなら、SQL を直接書く代わりに SQLAlchemy や SQLModel を使う方法もあります。

    WordPress のサイトと連携させる場合、WordPress 側から呼ぶなら wp_remote_get() や wp_remote_post() を使い、API キーは wp-config.php の定数などに置いてテーマのファイルに直接書かないようにします。

    まとめ

    • FastAPI は型ヒントと Pydantic のモデルから、入力検証と API ドキュメントを自動で作ります。
    • 作成用・更新用・返却用でモデルを分け、部分更新には exclude_unset=True を使います。
    • データベース接続は Depends で渡し、SQL の値はプレースホルダーで渡します。
    • 書き込みは API キーなどで制限し、キーは環境変数から読んで secrets.compare_digest() で比べます。
    • TestClient でテストを書き、本番は fastapi run か Uvicorn をリバースプロキシの後ろで動かします。
    Views: 35

    お気軽にご相談ください

    お見積りへ お問い合わせへ
    閉じる
    ▲