20200728のPythonに関する記事は30件です。

■Response

{
  "item_id": "1"
}

型宣言・バリデーション

指定した型ではないパラメータが渡されると、自動でエラーを返す

■Source

main.py

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return { "item_id": item_id }

■Request

http://localhost:8000/items/test

■Response

{
  "item_id": 1
}

■Request

■Response

{
  detail: [
    {
      loc: [
        "path",
        "item_id"
      ],
      msg: "value is not a valid integer",
      type: "type_error.integer"
    }
  ]
}

Path Operation Decoratorの順序

他のパスパラメータよりも優先したいURLがあった場合、順序に注意。
/users/meを後に書いてしまうと、パスパラメータが優先されてしまう。

■Source

main.py

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

■Request

http://localhost:8000/users/me

■Response

{
user_id: "the current user"
}

■Request

http://localhost:8000/users/user

■Response

{
user_id: "user"
}

パスパラメータの事前定義

Enumを利用することで有効なパスパラメータを事前定義することが可能。

■Source

main.py

from fastapi import FastAPI
from enum import Enum

app = FastAPI()

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/model/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}

    return {"model_name": model_name, "message": "Have some residuals"}```

■Request

http://localhost:8000/model/alexnet

■Response

{
  "model_name": "alexnet",
  "message": "Deep Learning FTW!"
}

http://localhost:8000/model/test

{
  detail: [
    {
      loc: [
        "path",
        "model_name"
      ],
      msg: "value is not a valid enumeration member; permitted: 'alexnet', 'resnet', 'lenet'",
      type: "type_error.enum",
      ctx: {
        enum_values: [
          "alexnet",
          "resnet",
          "lenet"
        ]
      }
    }
  ]
}

パスを含むパスパラメータ

パスパラメータにpathを記載することで、パスを受け取ることが可能。

■Source

main.py

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

■Request

http://localhost:8000/files/home/johndoe/myfile.txt

■Response

main.py

{
  "file_path": "/files/home/johndoe/myfile.txt"
}

Query Parameters

パスパラメータではないパラメータが渡された場合、自動的にクエリパラメータと判断される。

初期値

■Source

main.py

from fastapi import FastAPI

app = FastAPI()

fake_item_db = [
    {"name": "name1"},
    {"name": "name2"},
    {"name": "name3"},
    {"name": "name4"},
    {"name": "name5"},
    {"name": "name6"},
    {"name": "name7"},
    {"name": "name8"},
    {"name": "name9"},
    {"name": "name10"}
]

@app.get("/items")
async def read_item(skip: int = 0, limit: int = 3):
    return fake_item_db[skip : skip + limit]

■Request

http://localhost:8000/items?skip=3

■Response

[
  {
    name: "name1"
  },
  {
    name: "name2"
  },
  {
    name: "name3"
  }
]

■Request

■Response

[
  {
    name: "name4"
  },
  {
    name: "name5"
  },
  {
    name: "name6"
  }
]

■Request

http://localhost:8000/items?limit=5

■Response

[
  {
    name: "name1"
  },
  {
    name: "name2"
  },
  {
    name: "name3"
  },
  {
    name: "name4"
  },
  {
    name: "name5"
  }
]

■Request

http://localhost:8000/items?skip=3&limit=4

■Response

[
  {
    name: "name4"
  },
  {
    name: "name5"
  },
  {
    name: "name6"
  },
  {
    name: "name7"
  }
]

オプション

■Source

main.py

from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

■Request

http://localhost:8000/items/test

■Response

{
  "item_id": "test"
}

■Request

http://localhost:8000/items/test?q=aaa

■Response

{
  "item_id": "test",
  "q": "aaa"
}

型宣言

boolean型では1、True、on、yesという指定もtrueと判断される。

Source

main.py

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

■Request

http://localhost:8000/items/test

■Response

{
  "item_id": "test",
  "description": "This is an amazing item that has a long description"
}

■Request

http://localhost:8000/items/test?short=true

■Response

{
  "item_id": "test"
}

複数のパスとクエリパラメータ

■Source

main.py

@app.get("/users/{user_id}/items/{item_id}")
async def read_item(user_id: int, item_id: str, q: Optional[str] = None, short: bool = False):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

■Request

http://localhost:8000/users/1/items/test

■Response

{
  "item_id": "test",
  "owner_id": 1,
  "description": "This is an amazing item that has a long description"
}

必須のクエリパラメータ

■Source

main.py

@app.get("/items/{item_id}")
async def read_item(item_id: str, needy: str):
    item = {"item_id": item_id, "needy": needy}
    return item

■Request

http://localhost:8000/items/1?needy=test

■Response

{
  "item_id": "1",
  "needy": "test"
}

■Request

■Response

{
  detail: [
    {
      loc: [
        "query",
        "needy"
      ],
      msg: "field required",
      type: "value_error.missing"
    }
  ]
}

Request Body

データモデルによるリクエストボディ生成

■Source

main.py

from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

app = FastAPI()

@app.post("/items")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

■Request

{
  "name": "test",
  "price": 1000
}

■Response

{
  "name": "test",
  "description": null,
  "price": 1000,
  "tax": null
}

■Request

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

■Response

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100,
  "price_with_tax": 1100
}

リクエスト本文＋パスパラメータ

■Source

main.py

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}

■Request

http://localhost:8000/items/1?q=test

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

■Response

{
  "item_id": 1,
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

リクエスト本文＋パスパラメータ＋クエリパラメータ

■Source

main.py

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Optional[str] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

■Request

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

■Response

{
  "item_id": 1,
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100,
  "q": "test"
}

Query Parameters and String Validations

クエリパラメータに検証やメタデータを設定することが可能。

検証・オプションパラメータ

■Source

main.py

from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI()

@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, min_length=5, max_length=20)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

■Request

http://localhost:8000/items/?q=0123456789

■Response

{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "0123456789"
}

■Request

http://localhost:8000/items/?q=012345678901234567890

■Response

{
    detail: [
        {
            loc: [
                "query",
                "q"
            ],
            msg: "ensure this value has at most 20 characters",
            type: "value_error.any_str.max_length",
            ctx: {
                limit_value: 20
            }
        }
    ]
}

検証・必須パラメータ

■Source

main.py

@app.get("/items/")
async def read_items(q: str = Query(..., min_length=5, max_length=20)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    results.update({"q": q})
    return results

■Request

http://localhost:8000/items/?q=12345

■Response

{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "12345"
}

検証・正規表現

■Source

main.py

@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, min_length=5, max_length=20, regex="^test")):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

■Request

http://localhost:8000/items/?q=test12

■Response

{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "test12"
}

■Request

http://localhost:8000/items/?q=tes12

■Response

{
    detail: [
        {
            loc: [
                "query",
                "q"
            ],
            msg: "string does not match regex "^test"",
            type: "value_error.str.regex",
            ctx: {
                pattern: "^test"
            }
        }
    ]
}

リストのクエリパラメータ(デフォルトなし)

■Source

main.py

from fastapi import FastAPI, Query
from typing import List, Optional

app = FastAPI()

@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(None)):
    query_items = {"q": q}
    return query_items

■Request

http://localhost:8000/items/?q=123&q=456&q=test

■Response

{
  "q": [
    "123",
    "456",
    "test"
  ]
}

リストのクエリパラメータ(デフォルトあり)

■Source

main.py

from fastapi import FastAPI, Query
from typing import List, Optional

app = FastAPI()

@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(["foo", "bar"])):
    query_items = {"q": q}
    return query_items

■Request

■Response

{
  "q": [
    "foo",
    "bar"
  ]
}

メタデータの設定

設定したメタデータは自動でOpenAPIドキュメントへ反映される。
deprecatedをりようすることで非推奨であることを明示できる。
aliasを利用することでパスパラメータで受け取る際と、関数内の変数名を別にすることができる。

■Source

main.py

@app.get("/items/")
async def read_items(q: Optional[str] = Query(
        None,
        min_length=5,
        max_length=20,
        title="metadata_title",
        description="metadata_description",
        deprecated=True,
        alias="alias_q"
    )):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Path Parameters and Numeric Validations

メタデータの設定

パスパラメータでも同様にメタデータを設定することが可能。

■Source

main.py

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

app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
    item_id: int = Path(
      ...,
      title="metadata_title",
      description="metadata_description"
    ),
    q: Optional[str] = Query(
        None,
        min_length=5,
        max_length=20,
        title="metadata_title",
        description="metadata_description",
        deprecated=True,
        alias="alias_q"
    )):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

■Request

http://localhost:8000/items/1?q=test

■Response

{
  "item_id": 1
}

数値バリデーション

■Source

比較演算子	説明
ge	以上
gt	大きい
le	以下
lt	小さい

main.py

@app.get("/items/{item_id}")
async def read_items(*, item_id: int = Path(..., ge=1), q: str):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

■Request

■Response

{
  "item_id": 1,
  "q": "test"
}

Body - Multiple Parameters

パスパラメータ、クエリパラメータ、リクエストボディの組み合わせ

■Source

main.py

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

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),
    q: Optional[str] = None,
    item: Optional[Item] = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if item:
        results.update({"item": item})
    return results

■Request

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  }
}

複数のリクエストボディ

■Source

main.py

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class User(BaseModel):
    username: str
    full_name: Optional[str] = None

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

■Request

{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  }
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  }
}

リクエストボディの明示指定

パスパラメータではないパラメータを関数で指定すると、FastAPIはクエリパラメータと見なす。
クエリパラメータではなく、リクエストボディとしたい場合は明示的に= Body(...)を記載する。

■Source

main.py

from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class User(BaseModel):
    username: str
    full_name: Optional[str] = None

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

■Request

{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 0
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 0
}

複数のリクエストボディとクエリパラメータ

明示的なリクエストボディに加えて、クエリパラメータも指定可能。

■Source

main.py

from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class User(BaseModel):
    username: str
    full_name: Optional[str] = None

@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int,
    item: Item,
    user: User,
    importance: int = Body(..., gt=0),
    q: Optional[str] = None
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    if q:
        results.update({"q": q})
    return results

■Request

{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 1
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 1
}

Body - Fields

リクエストボディの各フィールドにメタタグを設定

スキーマにメタタグを設定することでOpenAPIへ自動反映される。

■Source

main.py

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

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None,
        title="The description of the item",
        max_length=300
    )
    price: float = Field(
        ...,
        gt=0,
        description="The price must be greater than zero"
    )
    tax: Optional[float] = None

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

Body - Nested Models

リクエストボディのフィールドにリストを設定

■Source

main.py

from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []

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

■Request

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [
    "foo",
    "bar"
  ]
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [
      "foo",
      "bar"
    ]
  }
}

ネストされたモデル

■Source

main.py

from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Image(BaseModel):
    url: str
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []
    image: Optional[Image] = None

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

■Request

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [
    "string"
  ],
  "image": {
    "url": "https://google.com",
    "name": "Google"
  }
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [
      "string"
    ],
    "image": {
      "url": "https://google.com",
      "name": "Google"
    }
  }
}

サブモデルのリスト

■Source

main.py

from typing import List, Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()

class Image(BaseModel):
    url: HttpUrl
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []
    images: Optional[List[Image]] = None

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

■Request

http://localhost:8000/offers/

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [
    "string"
  ],
  "images": [
    {
      "url": "https://google.com",
      "name": "Google"
    },
    {
      "url": "https://yahoo.com",
      "name": "Yahoo"
    }
  ]
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [
      "string"
    ],
    "images": [
      {
        "url": "https://google.com",
        "name": "Google"
      },
      {
        "url": "https://yahoo.com",
        "name": "Yahoo"
      }
    ]
  }
}

深くネストされたモデル

■Source

main.py

from typing import List, Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()

class Image(BaseModel):
    url: HttpUrl
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []
    images: Optional[List[Image]] = None

class Offer(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    items: List[Item]

@app.post("/offers/")
async def create_offer(offer: Offer):
    return offer

■Request

{
  "name": "string",
  "description": "string",
  "price": 0,
  "items": [
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    },
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    }
  ]
}

■Response

{
  "name": "string",
  "description": "string",
  "price": 0,
  "items": [
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    },
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    }
  ]
}

モデルのリスト

■Source

main.py

from typing import List
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()

class Image(BaseModel):
    url: HttpUrl
    name: str

@app.post("/images/multiple/")
async def create_multiple_images(images: List[Image]):
    return images

■Request

http://localhost:8000/images/multiple/

[
  {
    "url": "https://google.com",
    "name": "Google"
  },
  {
    "url": "https://yahoo.com",
    "name": "Yahoo"
  }
]

■Response

[
  {
    "url": "https://google.com",
    "name": "Google"
  },
  {
    "url": "https://yahoo.com",
    "name": "Yahoo"
  }
]

Schema Extra - Example

Configでスキーマ例の設定

■Source

main.py

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

    class Config:
        schema_extra = {
            "example": {
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            }
        }

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

モデルの各Fieldでスキーマ例の設定

■Source

main.py

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str = Field(..., example="Foo")
    description: Optional[str] = Field(None, example="A very nice Item")
    price: float = Field(..., example=35.4)
    tax: Optional[float] = Field(None, example=3.2)

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

関数でスキーマ例の設定

■Source

main.py

from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item = Body(
        ...,
        example={
            "name": "Foo",
            "description": "A very nice Item",
            "price": 35.4,
            "tax": 3.2,
        },
    ),
):
    results = {"item_id": item_id, "item": item}
    return results

Extra Data Types

■Source

main.py

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

app = FastAPI()

@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: Optional[datetime] = Body(None),
    end_datetime: Optional[datetime] = Body(None),
    repeat_at: Optional[time] = Body(None),
    process_after: Optional[timedelta] = Body(None),
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }

■Request

http://localhost:8000/items/3fa85f64-5717-4562-b3fc-2c963f66afa6

{
  "start_datetime": "2020-07-25T09:00:08.265Z",
  "end_datetime": "2020-07-25T09:00:08.265Z",
  "repeat_at": "18:20:30",
  "process_after": 0
}

■Response

{
  "item_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "start_datetime": "2020-07-25T09:00:08.265000+00:00",
  "end_datetime": "2020-07-25T09:00:08.265000+00:00",
  "repeat_at": "18:20:30",
  "process_after": 0,
  "start_process": "2020-07-25T09:00:08.265000+00:00",
  "duration": 0
}

Header Parameters

ヘッダーの利用

■Source

main.py

from typing import Optional
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}

■Request

http://localhost:8000/items/

■Response

{
  "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.89 Safari/537.36"
}

ヘッダーの重複

■Source

main.py

from typing import List, Optional
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(x_token: Optional[List[str]] = Header(None)):
    return {"X-Token values": x_token}

■Request

bash

curl -X GET "http://localhost:8000/items/" -H  "accept: application/json" -H  "x-token: foo,bar"

■Response

{
  "X-Token values": [
    "foo,bar"
  ]
}

Response Model

入力と異なる出力モデルを設定

■Source

main.py

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: Optional[str] = None

class UserOut(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user

■Request

http://localhost:8000/user/

{
  "username": "user",
  "password": "pass123",
  "email": "user@example.com",
  "full_name": "aiueo"
}

■Response

{
  "username": "user",
  "email": "user@example.com",
  "full_name": "aiueo"
}

デフォルト値の返却を除外

通常はモデルに指定したFieldすべてが返却される。
response_model_exclude_nusetを利用することで値が設定されたFieldのみ返却される。

■Source

main.py

from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

■Request

http://localhost:8000/items/foo

■Response

{
  "name": "Foo",
  "price": 50.2
}

返却するFieldを指定

モデルから返却するFieldを指定することが可能。

■Source

main.py

from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@app.get(
    "/items/{item_id}",
    response_model=Item,
    response_model_include={"name", "description"}
)
async def read_item(item_id: str):
    return items[item_id]

■Request

http://localhost:8000/user/

■Response

{
  "name": "Bar",
  "description": "The bartenders"
}

Extra Models

複数モデルの活用

■Source

main.py

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: Optional[str] = None

class UserOut(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

class UserInDB(BaseModel):
    username: str
    hashed_password: str
    email: str
    full_name: Optional[str] = None

def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

■Request

{
  "username": "string",
  "password": "string",
  "email": "string",
  "full_name": "string"
}

■Response

{
  "username": "string",
  "email": "string",
  "full_name": "string"
}

モデルの重複を排除

モデルの継承を利用することで重複を排除する。

■Source

main.py

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserBase(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

class UserIn(UserBase):
    password: str

class UserOut(UserBase):
    pass

class UserInDB(UserBase):
    hashed_password: str

def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

■Request

http://localhost:8000/user/

{
  "username": "string",
  "password": "string",
  "email": "string",
  "full_name": "string"
}

■Response

{
  "username": "string",
  "email": "string",
  "full_name": "string"
}

モデルのリストを返却

■Source

main.py

from typing import List
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str

items = [
    {"name": "Foo", "description": "There comes my hero"},
    {"name": "Red", "description": "It's my aeroplane"},
]

@app.get("/items/", response_model=List[Item])
async def read_items():
    return items

■Request

http://localhost:8000/items/

■Response

[
  {
    "name": "Foo",
    "description": "There comes my hero"
  },
  {
    "name": "Red",
    "description": "It's my aeroplane"
  }
]

Response Status Code

status_codeを利用することでHTTPステータスコードを設定することが可能。

ステータスコードの設定

■Source

main.py

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}

ステータスコードのショートカット

■Source

main.py

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

Form Data

フォームデータの受け取り

■Source

main.py

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

■Request

bash

curl -X POST "http://localhost:8000/login/" -H  "accept: application/json" -H  "Content-Type: application/x-www-form-urlencoded" -d "username=test&password=123"

Request Files

ファイルを送信

■Source

main.py

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"file": file}

@app.get("/")
async def main():
    content = """
    <html>
    <head>
      <title>FastAPI Form Test</title>
    </head>
    <body>
      <div>files</div>
      <form action="/files/" enctype="multipart/form-data" method="post">
        <input name="files" type="file">
        <input type="submit">
      </form>
      <div>uploadfiles</div>
      <form action="/uploadfiles/" enctype="multipart/form-data" method="post">
        <input name="files" type="file">
        <input type="submit">
      </form>
    </body>
    """
    return HTMLResponse(content=content)

■Request

bash

curl -X POST "http://localhost:8000/files/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "file=@img1.jpg;type=image/jpeg"

■Response

{
  "file_size": 386446
}

■Request

bash

curl -X POST "http://localhost:8000/uploadfile/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "file=@img2.jpg;type=image/jpeg"

■Response

{
  "file": {
    "filename": "img2.jpg",
    "content_type": "image/jpeg",
    "file": {
      "_file": {},
      "_max_size": 1048576,
      "_rolled": false,
      "_TemporaryFileArgs": {
        "mode": "w+b",
        "buffering": -1,
        "suffix": null,
        "prefix": null,
        "encoding": null,
        "newline": null,
        "dir": null,
        "errors": null
      }
    }
  }
}

複数ファイルを送信

■Source

main.py

from typing import List
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.post("/files/")
async def create_files(files: List[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

@app.get("/")
async def main():
    content = """
    <html>
    <head>
      <title>FastAPI Form Test</title>
    </head>
    <body>
      <div>files</div>
      <form action="/files/" enctype="multipart/form-data" method="post">
        <input name="files" type="file" multiple>
        <input type="submit">
      </form>
      <div>uploadfiles</div>
      <form action="/uploadfiles/" enctype="multipart/form-data" method="post">
        <input name="files" type="file" multiple>
        <input type="submit">
      </form>
    </body>
    """
    return HTMLResponse(content=content)

■Request

bash

curl -X POST "http://localhost:8000/files/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "files=@img1.jpg;type=image/jpeg" -F "files=@img2.jpg;type=image/jpeg"

■Response

{
  "file_sizes": [
    386446,
    320754
  ]
}

■Request

bash

curl -X POST "http://localhost:8000/uploadfiles/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "files=@img1.jpg;type=image/jpeg" -F "files=@img2.jpg;type=image/jpeg"

■Response

{
  "filenames": [
    "img1.jpg",
    "img2.jpg"
  ]
}

Request Forms and Files

フォームとファイルを同時に送信

■Source

main.py

from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()

@app.post("/files/")
async def create_file(
    file: bytes = File(...), fileb: UploadFile = File(...), token: str = Form(...)
):
    return {
        "file_size": len(file),
        "token": token,
        "fileb_content_type": fileb.content_type,
    }

■Request

bash

curl -X POST "http://localhost:8000/files/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "token=aaa" -F "file=@img1.jpg;type=image/jpeg" -F "fileb=@img1.jpg;type=image/jpeg"

■Response

{
  "file_size": 386446,
  "token": "aaa",
  "fileb_content_type": "image/jpeg"
}

Handling Errors

指定したエラーを返却

HTTPExceptionを利用することでステータスコード、メッセージを返却することが可能。

■Source

main.py

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

■Request

http://localhost:8000/items/foo

■Response

{
  "item": "The Foo Wrestlers"
}

■Request

http://localhost:8000/items-header/bar

■Response

{
  "detail": "Item not found"
}

カスタムヘッダーを追加

■Source

main.py

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},
        )
    return {"item": items[item_id]}

■Request

■Response

{
  "detail": "Item not found"
}

content-length: 27
content-type: application/json
date: Sun26 Jul 2020 02:47:20 GMT
server: uvicorn
x-error: There goes my error

カスタム例外ハンドラ

/unicorns/yoloにリクエストすると、raise UnicornExceptionとなる。
@app.exception_handler(UnicornException)としているため、実際にはこちらで処理が行われる。

■Source

main.py

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

■Request

http://localhost:8000/unicorns/yolo

■Response

{
  "message": "Oops! yolo did something. There goes a rainbow..."
}

content-length: 63
content-type: application/json
date: Sun26 Jul 2020 02:52:12 GMT
server: uvicorn

検証エラー時のリクエストボディ表示

検証エラーがあった際のResponseに送信されたリクエストボディを追加可能。

■Source

main.py

from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

class Item(BaseModel):
    title: str
    size: int

@app.post("/items/")
async def create_item(item: Item):
    return item

■Request

http://localhost:8000/items/

{
  "title": "towel",
  "size": "XL"
}

■Response

{
  "detail": [
    {
      "loc": [
        "body",
        "size"
      ],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ],
  "body": {
    "title": "towel",
    "size": "XL"
  }
}

Path Operation Configuration

応答ステータスコードの指定

status_codeで応答ステータスコードを指定することが可能。

■Source

main.py

from typing import Optional, Set
from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []

@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

OpenAPIドキュメントへの反映

タグ、サマリー、説明、レスポンスの説明、廃止状態を反映する。

■Source

main.py

from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []

@app.get(
    "/items/",
    tags=["items"],
    summary="summary test",
    description="description test",
    response_description="response test"
)
async def read_items():
    return [{"name":"Foo", "price":42}]

@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

@app.get("/users/", tags=["users"], deprecated=True)
async def read_users():
    return [{"username": "johndoe"}]

JSON Compatible Encoder

JSON互換のデータ型に変換

JSON互換のデータ型のみ格納可能なDBに対して、例えばdatetime型を格納する場合は文字列に変換する必要がある。
jsonable_encoderを利用するとJSON互換のデータに変換することが可能。

■Source

main.py

from datetime import datetime
from typing import Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

fake_db = {}

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = None

app = FastAPI()

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data

Body - Updates

PUT(置き換え)の注意

モデルが定義されたデータの置き換えを行う際に、初期値が設定されたFieldを省略すると、自動で初期値が格納される。

■Source

main.py

from typing import List, Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: float = 10.5
    tags: List[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_encoded = jsonable_encoder(item)
    items[item_id] = update_item_encoded
    return update_item_encoded

■Request

{
    "name": "Barz",
    "price": 3,
    "description": "description"
}

■Response

{
  "name": "Barz",
  "description": "description",
  "price": 3,
  "tax": 10.5,
  "tags": []
}

PATCHによる部分更新

exclude_unsetを利用することで、渡されたデータでモデルの部分更新を行う。

■Source

main.py

from typing import List, Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: float = 10.5
    tags: List[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item

■Request