FastAPI 入力検証 実践ガイド：クエリ/パス/ボディ/フォームを Annotated で型安全に、外部入力を境界で殺す

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(q: str | None = None):
    # q はクエリパラメータ。デフォルトが None なので「任意」。
    # ?q=foo で渡せば文字列、無ければ None。
    results = {"items": [{"item_id": "Foo"}]}
    if q:
        results.update({"q": q})
    return results

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(min_length=3, max_length=50)] = None,
):
    # 3〜50文字の文字列のみ受理。範囲外は 422 で自動的に弾かれる。
    # デフォルト値（= None）は関数引数側に置く。Query(default=...) は使わない。
    results = {"items": [{"item_id": "Foo"}]}
    if q:
        results.update({"q": q})
    return results

# pattern で「許される形」を正規表現で固定する（^...$ で全体一致を強制）。
q: Annotated[str | None, Query(min_length=3, max_length=50, pattern="^fixedquery$")] = None

# (1) 任意（省略可、既定 None）：来れば検証、無ければ None
q: Annotated[str | None, Query(min_length=3)] = None

# (2) 必須（省略不可）：デフォルトを書かない。省略すると 422
q: Annotated[str, Query(min_length=3)]

# (3) 必須だが None も明示で許す：デフォルトを書かない（＝必須）が、型は None も許容
q: Annotated[str | None, Query(min_length=3)]

# 必須クエリ：デフォルト値を書かない。省略すると 422。
@app.get("/items/")
async def read_items(q: Annotated[str, Query(min_length=3)]):
    return {"q": q}

from typing import Annotated
from fastapi import FastAPI, Path, Query

app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="取得するアイテムのID", ge=1)],
    q: Annotated[str | None, Query(alias="item-query")] = None,
):
    # item_id は 1 以上の整数のみ。0 や負数、文字列は 422。
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

# 1 ≦ item_id ≦ 1000 の整数。さらに 0 < size < 10.5 の float。
@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="アイテムID", ge=1, le=1000)],
    size: Annotated[float, Query(gt=0, lt=10.5)],
):
    return {"item_id": item_id, "size": size}

# q（デフォルトなし）の後に item_id を置いても、Annotated なら問題ない。
@app.get("/items/{item_id}")
async def read_items(
    q: str,
    item_id: Annotated[int, Path(title="アイテムID", ge=1)],
):
    return {"item_id": item_id, "q": q}

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: Annotated[list[str] | None, Query()] = None):
    # ?q=foo&q=bar → q == ["foo", "bar"]。リストには Query() が必要。
    return {"q": q}

# 何も来なければ ["foo", "bar"] を既定値として使う。
q: Annotated[list[str], Query()] = ["foo", "bar"]

# URL は ?item-query=foo、Python 側の変数は q。
q: Annotated[str | None, Query(alias="item-query")] = None

@app.get("/items/")
async def read_items(
    q: Annotated[
        str | None,
        Query(
            title="検索クエリ",
            description="アイテムを検索するための文字列",
            alias="item-query",
            min_length=3,
            max_length=50,
            deprecated=True,            # /docs に「非推奨」と表示（まだ動く）
            include_in_schema=False,    # OpenAPI スキーマから除外（docs に出さない）
        ),
    ] = None,
):
    return {"q": q}

from typing import Annotated, Literal
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()

class FilterParams(BaseModel):
    # Field でクエリごとの制約を宣言する。第8章で型エイリアスに畳んで再利用する。
    limit: int = Field(100, gt=0, le=100)        # 1〜100、既定 100
    offset: int = Field(0, ge=0)                 # 0 以上、既定 0
    order_by: Literal["created_at", "updated_at"] = "created_at"  # 列挙で固定
    tags: list[str] = []

@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
    # filter_query.limit / .offset / .order_by / .tags が検証済みで使える。
    return filter_query

class FilterParams(BaseModel):
    model_config = {"extra": "forbid"}           # 未知のクエリパラメータを禁止する
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []

{
    "detail": [
        {
            "type": "extra_forbidden",
            "loc": ["query", "tool"],
            "msg": "Extra inputs are not permitted",
            "input": "plumbus"
        }
    ]
}

from typing import Annotated
from fastapi import FastAPI, Path
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.put("/items/{item_id}")
async def update_item(
    item_id: Annotated[int, Path(title="アイテムID", ge=0, le=1000)],  # パス
    q: str | None = None,                                              # クエリ（単純型）
    item: Item | None = None,                                         # ボディ（モデル）
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if item:
        results.update({"item": item})
    return results

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    return {"item_id": item_id, "item": item, "user": user}

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

from fastapi import Body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: Annotated[int, Body(gt=0)],   # クエリではなくボディの値として、かつ > 0 を強制
):
    return {"item_id": item_id, "item": item, "user": user, "importance": importance}

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[Item, Body(embed=True)],   # item を "item" キーで包む
):
    return {"item_id": item_id, "item": item}

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = Field(
        default=None, title="アイテムの説明", max_length=300   # 最大300文字
    )
    price: float = Field(gt=0, description="価格はゼロより大きいこと")  # > 0 を強制
    tax: float | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    return {"item_id": item_id, "item": item}

from datetime import datetime, time, timedelta
from typing import Annotated
from uuid import UUID
from fastapi import Body, FastAPI

app = FastAPI()

@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,                                          # パスを UUID として検証・変換
    start_datetime: Annotated[datetime, Body()],           # ISO 8601 → datetime
    end_datetime: Annotated[datetime, Body()],
    process_after: Annotated[timedelta, Body()],           # 秒数 → timedelta
    repeat_at: Annotated[time | None, Body()] = None,
):
    # ここに来た時点で全部「検証済みの Python オブジェクト」。日時の演算がそのまま書ける。
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_process": start_process,
        "duration": duration,
    }

from enum import Enum
from fastapi import FastAPI

app = FastAPI()

class Status(str, Enum):
    draft = "draft"
    published = "published"
    archived = "archived"

@app.get("/items/")
async def read_items(status: Status = Status.draft):
    # ?status=foo は 422。?status=published は Status.published に変換される。
    return {"status": status}

from typing import Annotated
from fastapi import Query

# ページングと検索語の「契約」を1箇所に定義する（単一の真実源）。
Offset = Annotated[int, Query(ge=0, description="先頭から読み飛ばす件数")]
Limit = Annotated[int, Query(gt=0, le=100, description="取得する最大件数（上限100）")]
SearchQuery = Annotated[str | None, Query(min_length=2, max_length=50)]

@app.get("/items/")
async def list_items(q: SearchQuery = None, offset: Offset = 0, limit: Limit = 100):
    # 制約はエイリアスに集約。各ハンドラは名前を書くだけ。上限変更も1箇所で済む。
    return {"q": q, "offset": offset, "limit": limit}

@app.get("/users/")
async def list_users(offset: Offset = 0, limit: Limit = 100):
    # 同じページング契約を、users でもそのまま再利用する。
    return {"offset": offset, "limit": limit}

from pydantic import BaseModel

class ItemPublic(BaseModel):
    name: str
    price: float
    # internal_cost はあえて含めない → レスポンスに絶対出ない（型による情報遮断）

@app.get("/items/{item_id}", response_model=ItemPublic)
async def read_item(item_id: int) -> ItemPublic:
    # DB から内部フィールド付きで取得しても、ItemPublic に無いものは返らない。
    ...

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # 検証エラーを自社フロントの規約に合わせた一定の形へ整える。
    # exc.errors() に loc/msg/type が入る。生メッセージは出し過ぎない（情報漏洩を避ける）。
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "validation_error",
            "fields": [
                {"location": ".".join(str(p) for p in e["loc"]), "message": e["msg"]}
                for e in exc.errors()
            ],
        },
    )

# tests/test_validation.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

# 1) パスの数値制約：ge=1。0 や負数・文字列は 422、境界の 1 は 200。
def test_item_id_lower_bound():
    assert client.get("/items/1").status_code == 200       # 境界（OK）
    assert client.get("/items/0").status_code == 422       # 下限割れ
    assert client.get("/items/-3").status_code == 422
    assert client.get("/items/abc").status_code == 422     # 型変換失敗

# 2) クエリの文字列長：min_length=3。2文字は 422、3文字は 200。
def test_query_min_length():
    assert client.get("/items/?q=ab").status_code == 422   # 短すぎ
    assert client.get("/items/?q=abc").status_code == 200  # 境界（OK）

# 3) limit の上限：le=100。101 は 422、100 は 200。
def test_limit_upper_bound():
    assert client.get("/list/?limit=100").status_code == 200
    assert client.get("/list/?limit=101").status_code == 422

# 4) クエリモデルの extra=forbid：未知パラメータは 422 で拒否される。
def test_unknown_query_is_rejected():
    res = client.get("/items/?limit=10&unknown=x")
    assert res.status_code == 422
    assert res.json()["detail"][0]["type"] == "extra_forbidden"

# 5) 正常系：境界の内側はちゃんと通り、変換された値が返る。
def test_happy_path():
    res = client.get("/items/?q=hello&limit=10")
    assert res.status_code == 200