FastAPIとpydanticでREST APIを作る

Last updated at 2024-04-30Posted at 2024-04-07

はじめに

PythonでREST APIを構築する場合、候補となるフレームワークはいくつかありますが、代表的なものとしてはFastAPIとFlask（flask-smorest）があるかと思います。
私自身、2つのフレームワークを使って実際にAPIを作ったことはあるものの、その時はWebの記事の情報を断片的に収集して実装したこともあり、知識が定着していない状態でした。

そんなこんなで先日、「実践マイクロサービスAPI」の書籍でFastAPIとFlaskでREST APIを実装する実践的な方法を学んだので、本記事ではFastAPIを用いたAPIの実装手順をハンズオン形式で残そうと思います。

※別記事でFlask（flask-smorest）とmarshmallowを利用したREST APIの実装手順も投稿予定です。

前提

本記事では、FastAPIを使って最低限のAPI層（APIエンドポイントおよび入出力データの検証モデル）のみ実装します。

実際のAPI開発ではビジネスロジックやデータ層（DBとのデータの受け渡し）の実装も必要ですが、本記事では説明しません。

環境

OS

MacOS 14.4.1（23E224）

Python

python 3.10.3
fastapi==0.110.1
uvicorn==0.29.0
pydantic==2.6.4
pydantic[email]==2.6.4

本記事で使うサンプルのAPI仕様

以下の通り、一般的なユーザーの作成・更新・削除するAPIを作ります。

Method	Endpoint	説明	クエリパラメータ	リクエストボディ	レスポンスコード
POST	`/users`	指定した情報で新しいユーザーを作成する	-	`name`, `email`, `age`, `status`	`201`
GET	`/users`	ユーザーの一覧を取得する。クエリパラメータで取得する情報を制限することができる	`status` (オプション), `limit` (オプション)	-	`200`
GET	`/users/{userId}`	指定ユーザーの情報を取得する	-	-	`200`
PUT	`/users/{userId}`	指定ユーザーの情報を更新する	-	`name`, `email`, `age`, `status`	`200`
DELETE	`/users/{userId}`	指定ユーザーを削除する	-	-	`204`

API仕様の詳細は以下のopenapi.yamlの通りです。

openapi.yaml

openapi: 3.1.0
info:
  title: FastAPI
  version: 0.1.0
paths:
  "/users":
    get:
      summary: Get Users
      operationId: get_users_users_get
      parameters:
      - name: status
        in: query
        required: false
        schema:
          anyOf:
          - "$ref": "#/components/schemas/StatusEnum"
          - type: 'null'
          title: Status
      - name: limit
        in: query
        required: false
        schema:
          anyOf:
          - type: integer
          - type: 'null'
          title: Limit
      responses:
        '200':
          description: Successful Response
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/GetUsersSchema"
        '422':
          description: Validation Error
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/HTTPValidationError"
    post:
      summary: Create User
      operationId: create_user_users_post
      requestBody:
        required: true
        content:
          application/json:
            schema:
              "$ref": "#/components/schemas/CreateUserSchema"
      responses:
        '201':
          description: Successful Response
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/GetUserSchema"
        '422':
          description: Validation Error
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/HTTPValidationError"
  "/users/{user_id}":
    get:
      summary: Get User
      operationId: get_user_users__user_id__get
      parameters:
      - name: user_id
        in: path
        required: true
        schema:
          type: string
          format: uuid
          title: User Id
      responses:
        '200':
          description: Successful Response
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/GetUserSchema"
        '422':
          description: Validation Error
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/HTTPValidationError"
    put:
      summary: Update User
      operationId: update_user_users__user_id__put
      parameters:
      - name: user_id
        in: path
        required: true
        schema:
          type: string
          format: uuid
          title: User Id
      requestBody:
        required: true
        content:
          application/json:
            schema:
              "$ref": "#/components/schemas/CreateUserSchema"
      responses:
        '200':
          description: Successful Response
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/GetUserSchema"
        '422':
          description: Validation Error
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/HTTPValidationError"
    delete:
      summary: Delete User
      operationId: delete_user_users__user_id__delete
      parameters:
      - name: user_id
        in: path
        required: true
        schema:
          type: string
          format: uuid
          title: User Id
      responses:
        '204':
          description: Successful Response
        '422':
          description: Validation Error
          content:
            application/json:
              schema:
                "$ref": "#/components/schemas/HTTPValidationError"
components:
  schemas:
    CreateUserSchema:
      properties:
        user:
          "$ref": "#/components/schemas/UserSchema"
      additionalProperties: false
      type: object
      required:
      - user
      title: CreateUserSchema
    GetUserSchema:
      properties:
        user:
          "$ref": "#/components/schemas/UserSchema"
        id:
          type: string
          format: uuid
          title: Id
      additionalProperties: false
      type: object
      required:
      - user
      - id
      title: GetUserSchema
    GetUsersSchema:
      properties:
        users:
          items:
            "$ref": "#/components/schemas/GetUserSchema"
          type: array
          title: Users
      type: object
      required:
      - users
      title: GetUsersSchema
    HTTPValidationError:
      properties:
        detail:
          items:
            "$ref": "#/components/schemas/ValidationError"
          type: array
          title: Detail
      type: object
      title: HTTPValidationError
    StatusEnum:
      type: string
      enum:
      - active
      - inactive
      title: StatusEnum
    UserSchema:
      properties:
        name:
          type: string
          title: Name
        age:
          anyOf:
          - type: integer
            maximum: 120
            minimum: 0
          - type: 'null'
          title: Age
        email:
          type: string
          format: email
          title: Email
        status:
          allOf:
          - "$ref": "#/components/schemas/StatusEnum"
          default: active
      additionalProperties: false
      type: object
      required:
      - name
      - age
      - email
      title: UserSchema
    ValidationError:
      properties:
        loc:
          items:
            anyOf:
            - type: string
            - type: integer
          type: array
          title: Location
        msg:
          type: string
          title: Message
        type:
          type: string
          title: Error Type
      type: object
      required:
      - loc
      - msg
      - type
      title: ValidationError

データ検証モデル用のライブラリ

FastAPIは、データのバリデーションとシリアライゼーションのために内部的にpydanticを活用しています。このため、FastAPIを使用している場合、追加の設定なしでpydanticの全機能を直接利用することができます。

データ検証モデルとは？

APIのリクエスト/レスポンスのペイロードが期待する形式や条件を満たしていることを確認するためのルールを定めたものがデータ検証モデルである。
データ検証モデルを定義することで、リクエスト/レスポンスの入出力データがAPI仕様に合致しているか（期待するデータ型や値になっているか）をバリデーションすることができ、不備があればエラーを発生させることができる。これにより、不適切なデータがシステム内部に渡されるのを防ぐことでAPIの安全性と信頼性を高めることができる。

ディレクトリ構造

以下のディレクトリおよびファイルを作成します。

.
└──src
    └── FastAPI
        ├── api
        │   ├── api.py      # APIエンドポイント
        │   └── schemas.py   #　データ検証モデル
        └── app.py          # FastAPIアプリケーションの実行ファイル

FastAPIの実装手順

※以降は、「環境」に記載したライブラリ郡をpipインストールしていることが前提となります。

①FastAPI実行ファイルの実装

まずはFastAPIを実行できるようにFastAPIのインスタンスの生成とapiモジュールのインポートを行います。

FastAPI/app.py

from fastapi import FastAPI

# FastAPIクラスのインスタンスを生成
app = FastAPI(debug=True)

# apiモジュールをインポートとし、読み込み時にapi側のビュー関数（@app.get()等）を登録できるようにする
from src.FastAPI.api import api

②ユーザーAPIの最低限の実装

以下の通り、api.pyに最低限のエンドポイントを定義します。

FastAPI/api/api.py

from uuid import UUID

from starlette import status
from starlette.responses import Response

from src.FastAPI.app import app

# レスポンスを暫定的にスタブとして定義
user = {
        "id": "b3fafd48-cf8e-4c45-8323-b1963ed3a4f8",
        "user": {
            "name": "田中太郎",
            "email": "Cf9t8@example.com",
            "age": 20,
            "status": "active",
        }
}

@app.get("/users") # 1. HTTPメソッドに由来するデコレータとビュー関数
def get_users():
    return {"users": [user]}

@app.post("/users", status_code=status.HTTP_201_CREATED)
def create_user():
    return user

@app.get("/users/{user_id}") # 2. パスパラメータを指定
def get_user(user_id: UUID): # 3. 型ヒントで型を強制できる
    return user

@app.put("/users/{user_id}")
def update_user(user_id: UUID):
    return user

# 4. ステータスコードを指定すると上書きできる
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

ポイント

FastAPIでは、ビュー関数（get_users()など）に対してget()やpost()などのHTTPメソッドにちなんだ名前のデコレータを定義し、デコレータの引数内にエンドポイントのURLパスを登録する。
エンドポイントにパスパラメータ（{user_id}）を定義している場合、ビュー関数はuser_idという名前の引数を受け取る。
FastAPIでは型ヒントを使ってURLパスパラメータの型を指定できる。パスパラメータがその型に従ってない場合、リクエストは無効となる。
- user_idはUUID型を指定しているので、UUID型以外の型を渡した場合、FastAPIはエラーを返してくれる。

③FastAPIを実行してSwagger UIを確認する

src/FastAPIディレクトリに移動し、一旦この時点でFastAPIを実行してみます。

$ cd src/FastAPI

# FastAPIのホットリロード（ファイルを変更するたびに再起動する機能）を有効にして起動する
$ uvicorn src.FastAPI.app:app --reload
INFO:     Will watch for changes in these directories: ['/Users/src/FastAPI']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [6444] using StatReload
INFO:     Started server process [6446]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

その後、ブラウザでhttp://127.0.0.1:8000/docsにアクセスすると、api.pyに実装したコードからFastAPIが自動生成したAPIドキュメントの対話形式の画面（Swagger UI）が表示されます。
Swagger UIではAPI仕様を可視化できるだけでなく、各エンドポイントの実装をテストすることもできます。

試しに最上部の「GET /users」を展開し、Try it out　→ Executeを実行すると、api.pyに定義したスタブuserオブジェクトの値がレスポンスとして返却されることがわかります。

④リクエストペイロード用のデータ検証モデルを実装する

各APIエンドポイントのメインレイアウトが完成したので、ここからpydanticを使ってデータ検証モデルを作っていきます。
データ検証モデルを作ることで以下を実現します。

APIクライアントから受信したリクエストペイロードの検証
APIサーバが返却するレスポンスペイロードの検証

※正確には、ペイロードの検証が成功した後にデータのシリアライズ/デシリアライズも実行されます。

まずはAPIクライアントから受け取るリクエストペイロードを検証するためのモデルを作ります。
src/FastAPI/api/schemas.pyを以下のように実装します。

src/FastAPI/api/schemas.py

from enum import Enum
from typing import List, Optional

from pydantic import BaseModel, EmailStr, Extra, conint, validator

# ステータス型として列挙スキーマを定義
class StatusEnum(Enum):
    active = "active"
    inactive = "inactive"

# ユーザーのデータ検証モデルを定義する
# スキーマを定義するにはpydanticのBaseModelを継承したクラスを作る
class UserSchema(BaseModel):
    # 型ヒント(typing)を使って属性の型を指定する
    name: str
    # pydanticのconint関数を使ってageの数値範囲を制限し、strictで整数型（int）を強制する
    age: Optional[conint(ge=0, le=120, strict=True)]
    email: EmailStr
    # 型を列挙型にすることでプロパティで指定可能な値を制限できる
    status: StatusEnum = StatusEnum.active

    # ageはOptional型だが、Noneを許容しない
    @validator("age")
    def age_non_nullable(cls, value):
        assert value is not None, "age may not be None"
        return value

    # Configを使ってスキーマで定義されていないプロパティは許容しない
    class Config:
        extra = Extra.forbid

# POSTとPUTで受け取るリクエストペイロード用のデータ検証モデルを定義
class CreateUserSchema(BaseModel):
    user: UserSchema

    class Config:
        extra = Extra.forbid

ポイント

pydanticでは、プロパティにOptional型またはデフォルト値が指定されていない場合、そのプロパティは指定必須となる。
- name、emailは必須
- age、statusは省略可能
Optional型のフィールド（age）に対して"age": nullのリクエストを投げるとNoneがセットされてしまう。
- これを防ぐため、validatorメソッドで追加の検証ルールを追加し、「ageプロパティは含まれているが、値がnull（None）である場合はエラーを返す」バリデーションルールを追加する。

作成したデータ検証モデルをAPIエンドポイントに紐づける

api.pyのPOSTとPUTのビュー関数を以下のように変更します。

api.py

 from src.FastAPI.app import app
+# データ検証モデルをインポートする
+from src.FastAPI.api.schemas import CreateUserSchema
 
......

 @app.post("/users", status_code=status.HTTP_201_CREATED)
-def create_user():
+def create_user(user_details: CreateUserSchema):
     return {"user": user}

......
 
 @app.put("/users/{user_id}")
-def update_user(user_id: UUID):
+def update_user(user_id: UUID, user_details: CreateUserSchema):
     return {"user": user}

上記の通り、リクエストペイロードのデータ検証モデル（CreateUserSchema）をビュー関数の引数として宣言すると、ビュー関数はリクエストのペイロード（型や値）がデータ検証モデルに合致しているかどうかを自動的に判断し、合致しない場合はエラーを返してくれます。

データ検証モデルの動作確認

紐づけができたので、データ検証モデルが期待した通りに動作するかどうかをSwagger UIで確認します。
以降は「POST /users」タブを展開し、POSTのペイロードに変更を加えてリクエストを投げてみます。

必須項目を削除した場合

必須フィールドのnameフィールドを削除してExecuteを実行すると、次のようなエラーメッセージが表示されます。

{
  "detail": [
    {
      "type": "missing",
      "loc": [
        "body",
        "user",
        "name"  
      ],
      "msg": "Field required",
      "input": {
        "age": 0,
        "email": "user@example.com",
        "status": "active"
      },
      "url": "https://errors.pydantic.dev/2.6/v/missing"
    }
  ]
}

これはペイロードのどこでエラーが見つかったのかをFastAPIが自動生成してくれています。この例では"msg": "Field required"の通り、フィールドが不足している旨の
エラーが表示されています。

また、問題の箇所は以下のJSONポインタとして示されています。

"loc": [
        "body",
        "user",
        "name"  
      ]

JSONポインタの見方としては以下のとおりです。

enum型のフィールドに不正な値を入れた場合

statusはactiveまたはinactiveのいずれかの値を期待していますが、不正値を入れると以下のエラーが返却されます。

{
  "detail": [
    {
      "type": "enum",
      "loc": [
        "body",
        "user",
        "status"
      ],
      "msg": "Input should be 'active' or 'inactive'",
      "input": "fugafuga",
      "ctx": {
        "expected": "'active' or 'inactive'"
      }
    }
  ]
}

上記の通り、エラーメッセージの中で期待値も示してくれるので非常にわかりやすいですね。

データ検証モデルで定義されていないプロパティを送信した場合

ペイロードにタイプミスがある場合を想定し、emailをe-mailに変更してリクエストを投げると以下のエラーが返却されます。

{
  "detail": [
    {
      "type": "missing",
      "loc": [
        "body",
        "user",
        "email"
      ],
      "msg": "Field required",
      "input": {
        "name": "string",
        "age": 0,
        "e-mail": "user@example.com",
        "status": "active"
      },
      "url": "https://errors.pydantic.dev/2.6/v/missing"
    },
    {
      "type": "extra_forbidden",
      "loc": [
        "body",
        "user",
        "e-mail"
      ],
      "msg": "Extra inputs are not permitted",
      "input": "user@example.com",
      "url": "https://errors.pydantic.dev/2.6/v/extra_forbidden"
    }
  ]
}

Configクラスで指定したExtra forbidが効いていることがわかります。

Optional型にnull値が含まれている場合

ageプロパティにnullをセットしてリクエストを投げると以下のエラーが返却されます。

{
  "detail": [
    {
      "type": "assertion_error",
      "loc": [
        "body",
        "user",
        "age"
      ],
      "msg": "Assertion failed, age may not be None",
      "input": null,
      "ctx": {
        "error": {}
      },
      "url": "https://errors.pydantic.dev/2.6/v/assertion_error"
    }
  ]
}

validatorメソッドのチェックが効いていることを確認できます。

ここまでで、データ検証モデルで定義した型/条件に従ってリクエストペイロードのバリデーションが行われることをざっくり確認できました。

⑤レスポンスペイロードのデータ検証モデルの実装

ここからはAPIサーバが返却するレスポンスペイロードのデータ検証モデルを実装していきます。

schemas.pyの最下部にGetUserSchemaとGetUsersSchemaを追加します。

schemas.py

# 追加
from uuid import UUID
........

class CreateUserSchema(BaseModel):
    user: UserSchema

    class Config:
        extra = Extra.forbid

# 追加
class GetUserSchema(CreateUserSchema):
    id: UUID

# 追加
class GetUsersSchema(BaseModel):
    users: List[GetUserSchema]

GetUserSchemaクラス
- 本クラスは以下のエンドポイントが返すレスポンスペイロード用のデータ検証モデルとなる
  - GET　/user
  - POST /user
  - PUT /user
- 本クラスはCreateUserSchemaクラスを継承しているので、これはGetUserSchemaがCreateUserSchemaと同じ条件で検証されることを意味する
GetUserｓSchemaクラス
- 本クラスは以下のエンドポイントが返すレスポンスペイロード用のデータ検証モデルとなる
  - GET　/userｓ
- usersプロパティはList型であり、Listの要素としてはGetUserSchema型であることを意味する

レスポンスペイロードのデータ検証の必要性について

レスポンスペイロードはAPIサーバー側で生成されるため、その内容は基本的にはコントロール可能です。そのため、検証を絶対に必要とするわけではありませんが、以下の理由から検証を行うことが良いプラクティスとされています。

レスポンスペイロードを構成する元となるデータに誤りが含まれている可能性があります。例えば、データソースに不正な値が混入している場合などです。
ビュー関数のロジックを修正する過程で、誤ってレスポンスペイロードの一部を欠落させてしまう可能性があります。

このように、レスポンスペイロードの検証を行うことでデータの整合性を保ち、意図しないエラーやデータの不整合を防ぐことができます。

作成したデータ検証モデルをAPIエンドポイントに紐づける

api.pyを以下のように変更します。

api.py

 from src.FastAPI.app import app
-from src.FastAPI.api.schemas import CreateUserSchema
+from src.FastAPI.api.schemas import (
+    CreateUserSchema,
+    GetUsersSchema,  # 追加
+    GetUserSchema,   # 追加
+)
 
-@app.get("/users")
+@app.get("/users", response_model=GetUsersSchema)
 def get_users():
-    return {"users": [user]}
+    response = {"user": user}
+    return {"users": [response]}
 
-@app.post("/users", status_code=status.HTTP_201_CREATED)
+@app.post("/users",
+        status_code=status.HTTP_201_CREATED,
+        response_model=GetUserSchema
+)
 def create_user(user_details: CreateUserSchema):
     return {"user": user}
 
-@app.get("/users/{user_id}")
+@app.get("/users/{user_id}", response_model=GetUserSchema)
 def get_user(user_id: UUID):
     return {"user": user}
 
-@app.put("/users/{user_id}")
+@app.put("/users/{user_id}", response_model=GetUserSchema)
 def update_user(user_id: UUID, user_details: CreateUserSchema):
     return {"user": user}

上記のようにFastAPIでは、デコレータの引数response_modelにデータ検証モデルを指定することができます。これにより、レスポンスペイロードが指定したデータ検証モデルに準拠しているかどうかが確認されます。

データ検証モデルの動作確認

紐づけができたので、データ検証モデルが期待した通りに動作するかどうかを確認します。
確認方法としてはapi.py内にスタブとして定義したuserオブジェクトの値を変更し、Swagger UIでレスポンスを確認する流れになります。

まずはapi.pyのuserオブジェクトの値を以下のように変更します。

user = {
        "id": "b3fafd48-cf8e-4c45-8323-b1963ed3a4f8",
        "user": {
            # 必須パラメータ（name）を削除
            "email": "test",    # メールアドレス
            "age": 121,         # 範囲外の値を入れる
            "status": "テスト",   # enumで定義されていない値を入れる
            "hoge": "fuga",     # 不正なパラメータを追加する
        }
}

その後、Swagger UIから各エンドポイントに対してリクエストを発行し、一通りレスポンスを確認します。

Swagger UIでリクエストを送信すると500 ServerErrorが発生し、以下のバリデーションエラーが発生することを確認できます。

fastapi.exceptions.ResponseValidationError: 5 validation errors:
  {'type': 'missing', 'loc': ('response', 'user', 'name'), 'msg': 'Field required', 'input': {'id': '1111', 'email': 'test', 'age': 121, 'status': 'テスト', 'hoge': 'fuga'}, 'url': 'https://errors.pydantic.dev/2.6/v/missing'}
  {'type': 'less_than_equal', 'loc': ('response', 'user', 'age'), 'msg': 'Input should be less than or equal to 120', 'input': 121, 'ctx': {'le': 120}, 'url': 'https://errors.pydantic.dev/2.6/v/less_than_equal'}
  {'type': 'value_error', 'loc': ('response', 'user', 'email'), 'msg': 'value is not a valid email address: The email address is not valid. It must have exactly one @-sign.', 'input': 'test', 'ctx': {'reason': 'The email address is not valid. It must have exactly one @-sign.'}}
  {'type': 'enum', 'loc': ('response', 'user', 'status'), 'msg': "Input should be 'active' or 'inactive'", 'input': 'テスト', 'ctx': {'expected': "'active' or 'inactive'"}}
  {'type': 'extra_forbidden', 'loc': ('response', 'user', 'hoge'), 'msg': 'Extra inputs are not permitted', 'input': 'fuga', 'url': 'https://errors.pydantic.dev/2.6/v/extra_forbidden'}

また、ここでは説明は省略しますが、異常系とは別に正常なパラメータを与えたuserオブジェクトを作って動作を確認しておくとよいです。

FastAPIにおけるレスポンスペイロードの検証〜生成フロー

上記でレスポンスペイロードのデータ検証モデルを実装しましたが、FastAPIの内部処理としては以下のようになっています。

⑥インメモリでデータ保存できるようにする

これまでは、スタブとして定義したuserオブジェクトをレスポンスで返していましたが、この部分を変更して実際にリクエストペイロードのデータ保存と、保存されたデータを取り出してレスポンスを返せるようにします。保存先はシンプルにListにします。

インメモリのListでデータを保存できるようにするには、以下のようにapi.pyを変更する必要があります。
※変更箇所が多いので全文を載せます。

api.py

import uuid
from uuid import UUID

from fastapi import HTTPException
from starlette import status
from starlette.responses import Response

from src.FastAPI.app import app
from src.FastAPI.api.schemas import (
    CreateUserSchema,
    GetUsersSchema,
    GetUserSchema,
)

# 簡易的な保存用のインメモリのリスト
USERS = []

@app.get("/users", response_model=GetUsersSchema)
def get_users():
    return {"users": USERS}

@app.post("/users",
        status_code=status.HTTP_201_CREATED,
        response_model=GetUserSchema
)
def create_user(user_details: CreateUserSchema):
    # リクエストペイロードをdictにデシリアライズして扱えるようにする
    user = user_details.dict()
    user["id"] = uuid.uuid4()
    USERS.append(user)  # ユーザーをリストに追加（本来はDBに保存する）
    return user

@app.get("/users/{user_id}", response_model=GetUserSchema)
def get_user(user_id: UUID):
    for user in USERS:
        if user["id"] == user_id:
            return user

@app.put("/users/{user_id}", response_model=GetUserSchema)
def update_user(user_id: UUID, user_details: CreateUserSchema):
    for user in USERS:
        if user["id"] == user_id:
            user.update(user_details.dict())
            return user
    raise HTTPException(status_code=404, detail=f"User with ID {user_id} not found")

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: UUID):
    for index, user in enumerate(USERS):
        if user["id"] == user_id:
            USERS.pop(index)
            return Response(status_code=status.HTTP_204_NO_CONTENT)
    raise HTTPException(status_code=404, detail=f"User with ID {user_id} not found")

この状態でSwagger UIから「POST /users」にリクエストを投げると新しいユーザーがList内に保存され、作成されたユーザーの情報が以下のようにレスポンスとして返却されることが確認できます。

201レスポンス

{
  "user": {
    "name": "テストユーザー",
    "age": 22,
    "email": "user@example.com",
    "status": "active"
  },
  "id": "11c9ed63-cdc2-4437-8228-b42a2064240c"
}

２つ以上のユーザーを作成した後、「GET /users」にリクエストを投げると、作成したユーザー情報がリスト形式で返却されることが確認できます。

200レスポンス

{
  "users": [
    {
      "user": {
        "name": "テストユーザー１",
        "age": 0,
        "email": "user@example.com",
        "status": "active"
      },
      "id": "2ec6aeb5-94b9-4b64-a962-5ed839df2313"
    },
    {
      "user": {
        "name": "テストユーザー２",
        "age": 0,
        "email": "user@example.com",
        "status": "active"
      },
      "id": "7526f45d-ced3-4eea-8195-4e2362c03c14"
    }
  ]
}

また、「PUT /users/{user_id}」や「DELETE /users/{user_id}」のエンドポイントに対して存在しないIDをリクエストすると、以下のように404レスポンスが返ってくることも確認できます。

404レスポンス

{
  "detail": "User with ID 2fc4ba0d-c5f7-4991-bf61-bd1e998096a3 not found"
}

⑦`GET /users`エンドポイントにURLクエリパラメータを実装する

最後に、「GET /users」エンドポイントを拡張して以下２つのクエリパラメータを指定できるようにします。

status（任意項目）
- statusがactiveまたはinactiveいずれかの値を持つユーザーのみを取得することができる
limit（任意項目）
- 取得するユーザー情報を指定した数に限定することができる

ビュー関数にクエリパラメータを追加する

FastAPIではエンドポイントに対するクエリパラメータを簡単に実装できます。ビュー関数の引数にクエリパラメータを追加するだけです。
また、クエリパラメータ引数に型ヒントをつけることで検証ルールを追加することもできます。

api.py

+from typing import Optional

from src.FastAPI.api.schemas import (
     CreateUserSchema,
     GetUsersSchema,
     GetUserSchema,
+    StatusEnum
 )
 
@app.get("/users", response_model=GetUsersSchema)
-def get_users():
-    return {"users": USERS}
+# クエリパラメータは任意項目なのでOptionalをつける
+def get_users(status: Optional[StatusEnum] = None, limit: Optional[int] = None):
+    # パラメータが設定されていない場合はレスポンスをそのまま返す
+    if status is None and limit is None:
+        return {"users": USERS}
 
+    query_set = [user for user in USERS]
+
+    # statusの値をもとにクエリを絞り込む
+    if status is not None:
+        query_set = [
+            user
+            for user in query_set
+            if (user["status"] == status)
+        ]
+
+    # limitの値をもとにクエリを絞り込む
+    # limitが設定されている場合、その値がquery_setのサイズよりも小さい場合はquery_setのサブセットを返す
+    if limit is not None and len(query_set) > limit:
+        return {"users": query_set[:limit]}
+
+    return {"users": query_set}
+

この状態でSwagger UIの「GET /users」タブを開くと、以下のようにクエリパラメータが指定できるようになるのでそれぞれパラメータを入れて動作を確認します。

以上で簡単なAPIエンドポイントおよびデータ検証モデルの実装は完了です。
実際の開発ではこの後にビジネス層やデータ永続化のためのデータ層を作ることになると思います。

完成したコード

最後に作成したコードとopenapi.jsonを載せておきます。

app.py

from fastapi import FastAPI

app = FastAPI(debug=True)

from src.FastAPI.api import api

api.py

import uuid
from uuid import UUID
from typing import Optional

from fastapi import HTTPException
from starlette import status
from starlette.responses import Response

from src.FastAPI.app import app
from src.FastAPI.api.schemas import (
    CreateUserSchema,
    GetUsersSchema,
    GetUserSchema,
    StatusEnum
)

# 簡易的な保存用のインメモリのリスト
USERS = []

@app.get("/users", response_model=GetUsersSchema)
def get_users(status: Optional[StatusEnum] = None, limit: Optional[int] = None):
    if status is None and limit is None:
        return {"users": USERS}

    query_set = [user for user in USERS]

    if status is not None:
        query_set = [
            user
            for user in query_set
            if (user["status"] == status)
        ]

    if limit is not None and len(query_set) > limit:
        return {"users": query_set[:limit]}

    return {"users": query_set}

@app.post("/users",
        status_code=status.HTTP_201_CREATED,
        response_model=GetUserSchema
)
def create_user(user_details: CreateUserSchema):
    user = user_details.dict()
    user["id"] = uuid.uuid4()
    USERS.append(user)
    return user

@app.get("/users/{user_id}", response_model=GetUserSchema)
def get_user(user_id: UUID):
    for user in USERS:
        if user["id"] == user_id:
            return user

@app.put("/users/{user_id}", response_model=GetUserSchema)
def update_user(user_id: UUID, user_details: CreateUserSchema):
    for user in USERS:
        if user["id"] == user_id:
            user.update(user_details.dict())
            return user
    raise HTTPException(status_code=404, detail=f"User with ID {user_id} not found")

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: UUID):
    for index, user in enumerate(USERS):
        if user["id"] == user_id:
            USERS.pop(index)
            return Response(status_code=status.HTTP_204_NO_CONTENT)
    raise HTTPException(status_code=404, detail=f"User with ID {user_id} not found")

schemas.py

from enum import Enum
from typing import List, Optional
from uuid import UUID

from pydantic import BaseModel, EmailStr, Extra, conint, validator

class StatusEnum(Enum):
    active = "active"
    inactive = "inactive"

class UserSchema(BaseModel):
    name: str
    age: Optional[conint(ge=0, le=120, strict=True)]
    email: EmailStr
    status: StatusEnum = StatusEnum.active

    @validator("age")
    def age_non_nullable(cls, value):
        assert value is not None, "age may not be None"
        return value

    class Config:
        extra = Extra.forbid

class CreateUserSchema(BaseModel):
    user: UserSchema

    class Config:
        extra = Extra.forbid

class GetUserSchema(CreateUserSchema):
    id: UUID

class GetUsersSchema(BaseModel):
    users: List[GetUserSchema]

さいごに

本記事を通じて、FastAPIを使用したREST APIの基本的な実装手順を紹介しました。
FastAPIはPythonでAPIを構築する際の強力なフレームワークであり、その手軽さによって素早くAPIを開発することができます。

本記事では、ユーザーの作成、取得、更新、削除といった基本的なAPIエンドポイントの構築から、Pydanticを用いたリクエストとレスポンスのバリデーション、簡易的なインメモリデータベースを使用してのデータの保存と取得までを行いました。

ただ、ここで紹介した内容はFastAPIを使ったAPI開発の入門部分にすぎません。
実際のAPI開発では、ビジネスロジックの実装やデータベースとの連携、セキュリティ（認証と認可）など、考慮すべき要素がさらに多く存在します。

また、APIの設計や実装においては、仕様を明確にすることが非常に重要だと考えています。
FastAPIはOpenAPIスペックを自動生成する機能を持っており、この機能を活用することで、ドキュメントの整備とエンドポイントのテストを効率的に行うことができます。

次のステップとしては、別記事として実際のデータベースを使用したCRUD操作の実装や、認証機能の追加などを投稿してみようかなと思います！

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

FastAPIとpydanticでREST APIを作る

はじめに

前提

環境

OS

Python

本記事で使うサンプルのAPI仕様

データ検証モデル用のライブラリ

データ検証モデルとは？

ディレクトリ構造

FastAPIの実装手順

①FastAPI実行ファイルの実装

②ユーザーAPIの最低限の実装

③FastAPIを実行してSwagger UIを確認する

④リクエストペイロード用のデータ検証モデルを実装する

作成したデータ検証モデルをAPIエンドポイントに紐づける

データ検証モデルの動作確認

必須項目を削除した場合

enum型のフィールドに不正な値を入れた場合

データ検証モデルで定義されていないプロパティを送信した場合

Optional型にnull値が含まれている場合

⑤レスポンスペイロードのデータ検証モデルの実装

レスポンスペイロードのデータ検証の必要性について

作成したデータ検証モデルをAPIエンドポイントに紐づける

データ検証モデルの動作確認

FastAPIにおけるレスポンスペイロードの検証〜生成フロー

⑥インメモリでデータ保存できるようにする

⑦GET /usersエンドポイントにURLクエリパラメータを実装する

ビュー関数にクエリパラメータを追加する

完成したコード

さいごに

⑦`GET /users`エンドポイントにURLクエリパラメータを実装する