FastAPIを使ってみようと思うので、公式をざーっとみていく

ikepon

https://fastapi.tiangolo.com/ja/

ikepon

型: https://fastapi.tiangolo.com/ja/python-types/
- 定義しておくとエディタがいい感じに処理してくれたりエラー通知してくれる
- 型は実際にコード書きながら覚えればいいかな

ikepon

並行処理と async / await: https://fastapi.tiangolo.com/ja/async/
- 非同期対応の ORM とか外部 API 叩くとかは非同期にしておいた方が、待ち時間に他のリクエストとかを捌けるっぽい
- 非同期対応の ORM でない場合は遅くなったりするらしい
- async を定義しなくても使えるが、非同期対応してるなら定義して使った方がいい
- IOバウンドな処理はこれで良さそう
- CPUバウンドな処理は並列処理とマルチプロセッシングを使うと良いらしい
- FastAPI では async を path operation (クライアントからのリクエストを受ける関数のことだと思う、Rails でいう Controller 的なもの) で定義すればそれ以降はいい感じで解釈してくれるっぽい
  - FastAPIを使用している場合、その「最初の」関数がpath operation 関数であり、FastAPIが正しく実行する方法を知っているので、心配する必要はありません。
  - await を付けておけば大丈夫って感じかな
    - 関数が async def を使用して作成されている場合は、呼び出す際に await する必要があります。

ikepon

環境変数
- Python からは os.getenv で読み込める
- os.getenv("MY_NAME", "World") だと MY_NAME がある場合はそれを、なければ "World" を返すって意味だと思う
- 第二引数がなければ、 None が返るらしい
- The second argument to os.getenv() is the default value to return.
  If not provided, it's None by default, here we provide "World" as the default value to use.

ikepon

チュートリアルメモ

router
- ```
@app.get("/")
```
  - デコレータ、これの直下にこのメソッドとパスで処理する内容を書く
  - パスパラメータは {} で囲む
    - ex: @app.get("/items/{item_id}")
    - パラメータはそのまま item_id で使えそう
  - 直下の関数で型を設定できる
    - ex: async def read_item(item_id: int):
    - 型を宣言すると、FastAPI が自動的にその型に変換してくれる
- ```
async def root():
    return {"message": "Hello World"}
```
  - ここで処理する内容を書く
    - Rails でいう contorller のイメージだと思う
- パスパラメータの型指定は Enum でできる
  - ```
  class ModelName(str, Enum):
      alexnet = "alexnet"
      resnet = "resnet"
      lenet = "lenet"
  
  @app.get("/models/{model_name}")
  async def get_model(model_name: ModelName):
```
- 任意のパスをマルっと受け取る
  - ```
  @app.get("/files/{file_path:path}")
  async def read_file(file_path: str):
      return {"file_path": file_path}
```
- この方法だと file_path = '/aaa/bbb/ccc` とかを文字列としてマルっと受け取れる
- 型の path がそういう意味なんだと思う
- クエリパラメータ
  - async def root(): 部分で必要なパラメータと型を定義すれば自動的にクエリパラメータとして扱われる
  - オプショナルなパラメータは None にすればいい
    - ex: async def read_item(item_id: str, q: Union[str, None] = None):
    - この例だと q はオプショナルで、デフォルトは None
  - bool は FastAPI がいい感じで解釈してくれるっぽい
    - Trueと見なされる値: "True", "true", "on", "yes", "1"
    - Falseと見なされる値: "False", "false", "off", "no", "0"
    - Python 自体の bool 判定は None, False, 0. '', [], {} など
      - https://docs.python.org/3/library/stdtypes.html#truth-value-testing

ikepon

リクエストボディ
- Pydantic の BaseModel でモデル定義しておいて、それを引数の型で指定する
- 以下は公式の例

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel # ここで BaseModel をインポート


# モデル定義、これをそのまま引数の型に入れてる
class Item(BaseModel):
    name: str # これは必須になる
    description: Union[str, None] = None # None にしているのは任意
    price: float
    tax: Union[float, None] = None


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item): # 引数の型にして、リクエストボディで受け取れる
    return item

- これにより FastAPI がバリデーション、型変換、オブジェクト化などを自動でやってくれる

バリデーションを追加
- クエリパラメータにデフォルト値を設定しつつ、最小値、最大値、正規表現を設定

from typing import Union

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(
    q: Union[str, None] = Query(default=None, min_length=3, max_length=50, pattern="^fixedquery$"),
):
    # 必須の場合は q: str = Query(min_length=3, max_length=50, pattern="^fixedquery$") みたいにすれば良い
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

クエリパラメータのリスト型
- 以下で受け取れる
  - URL は ?q=string&q=string&q=string みたいになる
  - Railsだと ?q[]=string&q[]=string&q[]=string のイメージだったけど、FastAPI はサポートしてないので、 ?q=string&q=string&q=string の形で書かないといけないっぽい

from typing import List, Union

from fastapi import FastAPI, Query

app = FastAPI()


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

パスパラメータも Path を使って同様のことができそう
Query パラメータは Pydantic Model でも定義できる

from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


class FilterParams(BaseModel):
    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] = []


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
    return filter_query

query パラメータで不要なものを受けたくない場合は extra: forbid

class FilterParams(BaseModel):
    model_config = {"extra": "forbid"}

その他、いろいろあるけど必要に合わせて見れば良さそう

ikepon

リクエストボディ
- 一つだけ宣言するのは Query の場合と同じ
- 複数宣言すると、key が追加される
  - この辺は FastAPI がいい感じでやってくれてるっぽい

# 一つの場合
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = 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: Union[str, None] = None,
    item: Union[Item, None] = None,
):

属性を持つ JSON を期待する

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

# 複数持つとき
class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None


class User(BaseModel):
    username: str
    full_name: Union[str, None] = 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

このときは key に item, user を持つ JSON を期待する

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

また、上記のように key を持った JSON にしたい場合も Body を使えばよい

普通に宣言するとQueryパラメータと解釈されるが、リクエストボディに含めたい場合は Body を使う

from fastapi import Body, FastAPI

@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

このとき、key に importance が追加される

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

いろいろあるけど、この辺を実装の時に読み直す
- https://fastapi.tiangolo.com/ja/tutorial/body-multiple-params/#_5
バリデーションの設定
- https://fastapi.tiangolo.com/ja/tutorial/body-fields/#_1
ネストしたモデルの定義
- https://fastapi.tiangolo.com/ja/tutorial/body-nested-models/#_12
Schemaに例を追加
- https://fastapi.tiangolo.com/ja/tutorial/schema-extra-example/#pydanticschema_extra
その他の型
- https://fastapi.tiangolo.com/ja/tutorial/extra-data-types/#_3
クッキー
- https://fastapi.tiangolo.com/ja/tutorial/cookie-params/#cookie_1
リクエストヘッダー
- https://fastapi.tiangolo.com/ja/tutorial/header-params/#_4
クッキーの型宣言
- https://fastapi.tiangolo.com/ja/tutorial/cookie-param-models/#forbid-extra-cookies
ヘッダーの型宣言
- https://fastapi.tiangolo.com/ja/tutorial/header-param-models/#forbid-extra-headers

ikepon

レスポンスボディ
- response_model で定義
- パスワードなど、リクエストでは受け取ってレスポンスには含まないものは別モデルで定義する

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None


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

デフォルト値を設定した場合、レスポンスに含まれてしますが、これを除きたい場合は response_model_exclude_unset=True

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

response_model_exclude_none=True で None を除けるっぽい
困ったらここを見る
- https://fastapi.tiangolo.com/ja/tutorial/response-model/#_8

ikepon

モデル
- IN と OUT で異なる部分がある場合は XxxBase モデルを作って継承させる
  - https://fastapi.tiangolo.com/ja/tutorial/extra-models/#_2

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None


class UserInDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: Union[str, None] = 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

ステータスコード
- status_code で定義

@app.post("/items/", status_code=201)

# FastAPI の status を使える
from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)

ikepon

フォームデータ
- https://fastapi.tiangolo.com/ja/tutorial/request-forms/#_2
- https://fastapi.tiangolo.com/ja/tutorial/request-form-models/#summary
- multipart は以下が必要
  - https://github.com/Kludex/python-multipart
File と UploadFile
- 基本的には UploadFileを使ってれば大丈夫そう
- ファイルが小さくて、それを読み込んですぐ使いたい場合は File で良い

ikepon

エラーハンドリング
- raise HTTPException(status_code=404, detaiil="message") を使う
- エラーなので raise
- 例外(エラー)が発生した場合、それ以降のコードは実行されない

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]}

この例だと status code 404、レスポンスボディは以下が返る

{
  "detail": "Item not found"
}

detail
- dist や list を渡せる
カスタム例外ハンドラー
- @app.exception_handler() で例外をキャッチできる
以下の例だと、custom例外で UnicornException を定義して、それを必要箇所で raise する
- @app.exception_handler(UnicornException) で例外をキャッチ
- 例外を受け取って、処理
- って流れだと思う

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}

ikepon

デフォルトエラー
- RequestValidationError
  - リクエストに無効なデータが含まれている場合に発生する
- これをカスタムしたい場合は @app.exception_handler() で拾って上書きすればいい

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)

その他、エラーの処理方法がいろいろあるので必要に合わせて読む

ikepon

Path Operationの設定
- https://fastapi.tiangolo.com/ja/tutorial/path-operation-configuration/
tags パラメータ
- OpenAPI の tags に追加できる
  - これ、クエリパラメータとかと重複するのでは？
    - -> @app の方だから大丈夫
OpenAPI のドキュメントにどう細かいところを追加するかの説明だな

ikepon

JSON 互換エンコーダ
- https://fastapi.tiangolo.com/ja/tutorial/encoder/
- データ型(Pydanticみたいな)を JSON と互換性のあるもの(dict, list など)に変換する
- jsonable_encoder
更新
- PUT はマルっと置き換えるので、リクエストボディで値がないものがあれば、デフォルト値に置き換わってしまう
- PATCH で dict(exclude_unset=True) を設定すれば、リクエストボディに含まれる値のみ更新がかかる
  - Rails だと当たり前にやってくれてたことだけど、この辺の差分は理解しとかないとだな
  - .copy で既存モデルのコピーを作っておいて、update で更新するデータを入れることができる

from typing import List, Union

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: Union[str, None] = None
    description: Union[str, None] = None
    price: Union[float, None] = 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

上記フロー
- PATCH でリクエスト
- 既存データを取得 -> Pydantic モデルに変換
- リクエストボディを dict(exclude_unset=True) で変換
- 先ほど作ったPydanticモデルを copy して、そこに更新したい値を update で上書きする
- それを DB 保存できるものに変換(jsonable_encoder)
- DB 保存