FastAPI: 自動バリデーション付きREST APIの構築
1. 概要
REST APIを書く際に、防御的なコードの大部分は入力のチェックにあります。フィールドが必須かどうか、数が負でないか、メールアドレスが有効かどうか。FastAPIはこれをPydanticに委譲します。型を宣言して制約を指定するだけで、フレームワークが自動的にリクエストをバリデーションし、構造化された422エラーを返します。
このチュートリアルでは、最初から注文管理API(Order)を構築します。読み終わったら、body、path、queryを制約する方法を知ることができます。カスタムバリデーターを書き、返すデータをフィルタリングし、エラー形式を変更し、pytestで完全にテストする方法も習得します。
2. インストール
Python 3.10以上が必要です。virtualenvを作成してから、FastAPIをインストールしてください(standard依存関係グループを含めて):
python -m venv .venv
source .venv/bin/activate
pip install "fastapi[standard]" pytest
パッケージfastapi[standard]はこのチュートリアルに必要なものをすべて含んでいます:
uvicornサーバーを実行するため。fastapi-cliコマンドfastapi devを提供する。httpxTestClient用。email-validatorEmailStr型用。
3. Pydanticモデルを使った最初のエンドポイント
ファイルmain.pyを作成してください:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="Order API")
class OrderCreate(BaseModel):
customer_email: str
product_sku: str
quantity: int
@app.post("/orders", status_code=201)
def create_order(order: OrderCreate) -> OrderCreate:
return order
開発モードでサーバーを実行(自動リロード機能付き):
fastapi dev main.py
ここではまだif文がありません。パラメータorder: OrderCreateはFastAPIに3つのことを指示します:
- リクエストボディからJSONを読む。
- そのJSONを
OrderCreateにパースする。 - データが型と一致しない場合は
422を返す。
quantityを数値ではない文字列として送信してみてください:
curl -X POST localhost:8000/orders \
-H "Content-Type: application/json" \
-d '{"customer_email":"an@example.com","product_sku":"TSHIRT-01","quantity":"abc"}'
{"detail":[{"type":"int_parsing","loc":["body","quantity"],
"msg":"Input should be a valid integer, unable to parse string as an integer",
"input":"abc"}]}
locフィールドはエラーの正確な場所を示します:body内のquantityフィールド。さらに、/docsのOpenAPIドキュメントはこのモデルから生成されます。したがってコードとドキュメントは同期しています。
4. Fieldによるデータ制約
型チェックだけでは不十分です。例えばquantity = -5は依然として有効なintです。Fieldを使って制約を追加します(Annotatedと組み合わせて):
from decimal import Decimal
from typing import Annotated
from pydantic import BaseModel, EmailStr, Field
Sku = Annotated[str, Field(min_length=3, max_length=20, pattern=r"^[A-Za-z0-9-]+$")]
class OrderCreate(BaseModel):
customer_email: EmailStr
product_sku: Sku
quantity: Annotated[int, Field(gt=0, le=100)]
unit_price: Annotated[Decimal, Field(gt=0, max_digits=10, decimal_places=2)]
discount: Annotated[Decimal, Field(ge=0, decimal_places=2)] = Decimal("0")
note: Annotated[str | None, Field(max_length=200)] = None
各行は次の操作を行います:
EmailStrは実際のメール形式をチェックします(@文字があるかどうかをチェックするだけではなく)。Skuは再利用可能な型です。3~20文字の長さが必須で、文字、数字、ハイフンのみを含みます。quantityは0より大きく、100を超えない必要があります。unit_priceはDecimalを使います(floatではなく)。金額を計算する際に丸め誤差を避けるためです。最大10桁、小数点以下2桁。discountとnoteはデフォルト値を持つオプショナルフィールドです。デフォルト値のないフィールドは必須フィールドです。
quantity: 0を送信すると、次が返されます:
{"type":"greater_than","loc":["body","quantity"],"msg":"Input should be greater than 0","input":0,"ctx":{"gt":0}}
注記: デフォルトではPydanticはlaxモードで実行されます。文字列
"5"は数値5に強制変換されます(エラーを報告せずに)。完全に拒否したい場合はmodel_config = ConfigDict(strict=True)をモデルに追加するか、各フィールド用にField(strict=True)を使用してください。
5. pathおよびqueryパラメータのバリデーション
bodyだけでなく、pathとqueryも同じメカニズムでバリデーションされます:
from fastapi import HTTPException, Path, Query
orders_db: dict[int, dict] = {}
@app.get("/orders/{order_id}")
def get_order(order_id: Annotated[int, Path(gt=0)]):
if order_id not in orders_db:
raise HTTPException(status_code=404, detail="Order not found")
return orders_db[order_id]
@app.get("/orders")
def list_orders(
page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(ge=1, le=50)] = 10,
sku: Annotated[str | None, Query(min_length=3)] = None,
):
items = [o for o in orders_db.values() if sku is None or o["product_sku"] == sku]
start = (page - 1) * size
return items[start : start + size]
異なるリクエストの結果:
GET /orders/0は422を返します(order_idは0より大きい必要があるため)。GET /orders?size=500は422を返します(sizeの最大値は50のため)。GET /orders?page=2&size=5は正常に実行されます。pageとsizeの2つの値は関数に入る前にintに強制変換されました。
422と404を区別してください。422エラーはリクエストの形式が間違っていることを意味します。404エラーはビジネスロジックなので、自分でraise HTTPExceptionする必要があります。
6. カスタムバリデーター
Fieldで表現できないルールがあります。その場合、単一フィールド用にfield_validatorを、複数フィールドに関する規則用にmodel_validatorを使用します。
from pydantic import field_validator, model_validator
FORBIDDEN_DOMAINS = {"mailinator.com", "tempmail.com"}
class OrderCreate(BaseModel):
# ... 前のセクションのフィールド ...
@field_validator("product_sku")
@classmethod
def normalize_sku(cls, value: str) -> str:
return value.upper()
@field_validator("customer_email")
@classmethod
def reject_disposable_email(cls, value: str) -> str:
domain = value.split("@")[-1].lower()
if domain in FORBIDDEN_DOMAINS:
raise ValueError("Disposable email is not allowed")
return value
@model_validator(mode="after")
def check_discount(self) -> "OrderCreate":
total = self.unit_price * self.quantity
if self.discount > total:
raise ValueError("Discount cannot exceed order total")
return self
各バリデーターを説明します:
normalize_skuはFieldがパターンをチェックした後に実行されます。SKUを大文字に正規化するので、"tshirt-01"は"TSHIRT-01"として保存されます。バリデーターは値を変換できます(新しい値を返す限り)。reject_disposable_emailはValueErrorを発生させます。Pydanticはこれを検証エラーに変換し、メッセージに"Value error, "というプレフィックスが付きます。check_discountはmode="after"で実行されます。つまり、すべてのフィールドが有効になり正しい型を持つ時点で実行されます。これにより、Decimalを安全に乗算できます。return selfを忘れずに。
注記: バリデーター内からデータベースや外部APIを呼び出さないでください。例えば、「SKUが在庫に存在するか」をチェックするのはビジネスロジックです。それをserviceまたはdependencyに配置してください。バリデーターは純粋な関数であるべき、速く、テストしやすい。
7. response_model: 出力データの制御
通常、clientに返すレベルより多くの情報を保存します(例えば内部メモ)。入出力モデルを分離してください:
class OrderRead(BaseModel):
id: int
customer_email: EmailStr
product_sku: str
quantity: int
total: Decimal
@app.post("/orders", status_code=201, response_model=OrderRead)
def create_order(order: OrderCreate):
order_id = len(orders_db) + 1
record = {
"id": order_id,
**order.model_dump(),
"total": order.unit_price * order.quantity - order.discount,
"internal_flag": "manual-review",
}
orders_db[order_id] = record
return record
関数はnoteとinternal_flagの両方を含むdictを返します。しかし、response_model=OrderReadはレスポンスをフィルタリングし、宣言されたフィールドのみを返します:
{"id":1,"customer_email":"an@example.com","product_sku":"TSHIRT-01","quantity":2,"total":"39.98"}
2つの点に注意する必要があります:
Decimalは文字列"39.98"にシリアル化されます(精度を保つために)。クライアントはこれを知っている必要があります。- レスポンスもバリデーションされます。
totalフィールドを忘れた場合、FastAPIはResponseValidationErrorをスロー、500を返します(422ではなく)。これはサーバーエラー(クライアントエラーではなく)。
8. 422エラーの形式をカスタマイズ
デフォルト形式は非常に詳細です。多くのチームはよりコンパクトで、フロントエンドで統一された構造を望みます。ハンドラーをオーバーライドできます:
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = [
{
"field": ".".join(str(part) for part in err["loc"][1:]) or "body",
"message": err["msg"],
}
for err in exc.errors()
]
return JSONResponse(
status_code=422,
content={"code": "VALIDATION_FAILED", "errors": errors},
)
ハンドラーが各エラーを処理する方法:
err["loc"][1:]は最初の要素(body、query、path)を削除し、残りを点で結合します。model_validatorからのエラーはどのフィールドにも関連付けられていません。この場合、locは("body",)のみなので、ハンドラーは"body"を割り当てます。
quantity: 0と一度限りのメールアドレスを持つオーダーを送信すると、次が返されます:
{"code":"VALIDATION_FAILED","errors":[
{"field":"customer_email","message":"Value error, Disposable email is not allowed"},
{"field":"quantity","message":"Input should be greater than 0"}]}
Pydanticは最初のエラーで停止する代わりに、一度に複数フィールドからすべてのエラーを収集します。クライアントは1ラウンドで全て修正できる。
9. pytestでのテスト
TestClientはアプリを直接呼び出します(サーバーを実行する必要がない)。ファイルtest_main.pyをmain.pyの隣に作成してください:
import pytest
from fastapi.testclient import TestClient
from main import app, orders_db
client = TestClient(app)
VALID_ORDER = {
"customer_email": "an@example.com",
"product_sku": "tshirt-01",
"quantity": 2,
"unit_price": "19.99",
}
@pytest.fixture(autouse=True)
def clear_db():
orders_db.clear()
def test_create_order_returns_normalized_sku_and_total():
response = client.post("/orders", json=VALID_ORDER)
assert response.status_code == 201
body = response.json()
assert body["product_sku"] == "TSHIRT-01"
assert body["total"] == "39.98"
assert "internal_flag" not in body
def test_invalid_quantity_and_email_are_reported_together():
payload = {**VALID_ORDER, "quantity": 0, "customer_email": "x@mailinator.com"}
response = client.post("/orders", json=payload)
assert response.status_code == 422
fields = {e["field"] for e in response.json()["errors"]}
assert fields == {"quantity", "customer_email"}
def test_discount_greater_than_total_is_rejected():
payload = {**VALID_ORDER, "discount": "100.00"}
response = client.post("/orders", json=payload)
assert response.status_code == 422
assert response.json()["errors"][0] == {
"field": "body",
"message": "Value error, Discount cannot exceed order total",
}
def test_page_size_over_limit_is_rejected():
assert client.get("/orders?size=500").status_code == 422
テストを実行してください:
pytest -q
.... [100%]
4 passed in 0.31s
フィクスチャーclear_dbは各テストの前にorders_dbをリセットし、テストが互いに独立している。さらに、HTTPなしでモデルを個別にテストできます。OrderCreate.model_validate(...)を使用し、pytest.raises(ValidationError)と組み合わせてください。この方法はモデルが複雑な多くのルールを持つ場合に適切です。
10. よくあるエラー
floatを金額に使う。0.1 + 0.2は0.3に等しくありません。Decimalを使ってください(max_digitsとdecimal_places付き)。@classmethodを忘れるか、デコレーターの順序が間違う。@field_validatorは@classmethodの上にある必要があります。そうしないと、Pydantic v2はimport時にエラーを報告します。- バリデーターで
returnを忘れる。field_validatorが返さない場合、フィールドにNoneが割り当てられます。model_validator(mode="after")はselfを返す必要があります。 - Pydantic v1構文を使用。
@validator、class Config、.dict()は非推奨です。@field_validator、model_config、.model_dump()を使用してください。 422と500を混同。 不正な入力は422を出す。レスポンスがresponse_modelと一致しない場合は500を出す。500が表示されたら、返すデータを確認。- 入出力に同じモデルを使用。 これにより、内部フィールドが公開されやすくなり、クライアントに
idなどのフィールドを送信させます。OrderCreateとOrderReadを分離してください。
注記: API層のバリデーションはデータベースの制約に代わるものではありません。
UNIQUEまたはNOT NULLなどの制約はまだ必要です。リクエストはシステムに異なる複数の経路から入る可能性があります(例えばバックグラウンドジョブまたは移行スクリプト)。
11. 結論
FastAPIはPydanticと組み合わせて、型ヒントをデータコントラクトに変えます。3つの主な利点があります:
- 不正なリクエストはエントリーポイントで遮断されます。
- エラーは完全に構造化されて返されます。
- OpenAPIドキュメントは常にコードと同期します。
このチュートリアルでは、Fieldをシンプルな制約用に、PathとQueryをURLパラメータ用に、field_validatorとmodel_validatorをカスタムルール用に使いました。response_modelを使って出力をフィルタリングし、422エラー形式を変更し、TestClientで完全にテストしました。
覚えておくべき原則:バリデーターにデータの形状をチェックさせ、ビジネスロジックはサービス層に配置してください。