FastAPI APIを使ってみる

目的

WindowsPCでVscodeを使ってFAST APIを少し触ってみました。
公式のチュートリアルを読解して、一部は動作確認をしました。

自分用メモとしてまとめました。
間違っている可能性があることをご了承ください。

実施内容

Python環境構築

公式からインストールする。
https://www.python.org/downloads/

インストール時に下記のチェックを忘れない。

適当なフォルダを作成してVscodeで開く。

Vscode の拡張機能からPythonをインストールする。

適当なtest.pyを作成して実行してみる。
単純なファイルは右上の▶で実行できる。

FAST APIを構築

公式：https://fastapi.tiangolo.com/ja/

公式のトップページを参照して下記二つのコマンドを実行する。

$ pip install fastapi
$ pip install uvicorn

main.pyを公式の指示通り作成して$ uvicorn main:app --reloadを実行する。

http://127.0.0.1:8000/items/10?q=somequeryなどでAPIを実行したり、
http://127.0.0.1:8000/docsからAPI定義がswaggerで閲覧できる。

おまけ：Python Types Intro

tuple、dict、Union、Optional、Pydantic、Annotatedとか記載してあります。
https://fastapi.tiangolo.com/ja/python-types/#__tabbed_4_2

FAST APIのチュートリアル

最初のステップ

FastAPIの使い方、APIの基本が記載されている。

パスパラメータ

型チェック、パスの評価順序、列挙型を用いたAPIの挙動の説明。

パラメータにPathを含んでも対応できる。
例えば下記のAPIを実行するとき、
http://127.0.0.1:8000/files/aaaなどは通常通り機能するが、
http://127.0.0.1:8000/files/C:/Users/test/aaa.txtでも機能する。

.py

from fastapi import FastAPI

app = FastAPI()


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

C:/Users/test/aaa.txtを1つのgetパラメータとして処理してくれる。

クエリパラメータ

パスパラメータやクエリパラメータの基本的な使い方の説明。
許容する型の指定、必須の設定など。

boolean方のクエリパラメータであるshortを指定するとき、
short=trueやshort=1でtrueはもちろん、
short=yesやshort=onでもshor=trueとして解釈してくれるのは意外でしてた。

リクエストボディ

pydanticを用いたリクエストボディの説明。

.py

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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

クエリパラメータと文字列の検証

Queryを用いたクエリパラメータのバリデーション(最大文字数、初期値、必須など)。
またQueryを用いれば、titleやdescriptionやaliasなどの情報を付加することもできる。

.py

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,
        title="Query string",
        description="Query string for the items to search in the database that have a good match",
        min_length=3,
        alias="item-query"
    )
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Path Parameters and Numeric Validations

AnnotatedとPathを用いれば、引数に対して数値のバリデーションなどができる。

下記の例だとge=1で1以上の値という制約をつけている。

.py

from typing import Annotated

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get", ge=1)], q: str
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

Body - Multiple Parameters

1つのリクエストに対して2つのオブジェクト(ItemとUser)を含む場合は下記のように記載する。

.py

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


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


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

リクエストは下記のように記載する。

.py

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

また、1つのリクエストに対してオブジェクトが1つの場合でもBody(embed=True)を指定することでオブジェクトのキーをリクエストに要求することができる。
↓Body(embed=True)あり

.py

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

↓Body(embed=True)なし

.py

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

Body - Fields

AnnotatedとFieldを用いれば、クラスに対して数値のバリデーションなどができる。

main.py

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

Body - Nested Models

入れ子になっているクラスやリストの説明。オブジェクト指向がわかれば問題ない。
pydanticのhttpurlの説明。

Declare Request Example Data

PathやFieldなどのexamplesを用いることでサンプル値を設定することができる。
defaultよりも優先的に適用される。

複数のexamplesについて、Swagger UI上にチュートリアル通り表示されない。
OpenAPI 3.1.0が原因かも。

Extra Data Types

strやint以外にもUUIDやdatetimeなどの方も使用できるという説明。

クッキーのパラメータ

Queryと同様の使用方法でCookieを用いれば、クッキーのパラメータもバリデーションなどが可能。

.py

from typing import Union

from fastapi import Cookie, FastAPI

app = FastAPI()


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

ヘッダーのパラメータ

Cookieと同様の使用方法でHeaderを用いれば、ヘッダーのパラメータもバリデーションなどが可能。

.py

from typing import List, Union

from fastapi import FastAPI, Header

app = FastAPI()


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

デフォルトの設定ではアンダーバーはハイフンに自動変換される。
この変換を無効にする場合はconvert_underscoresをFalseに設定する。

Response Model - Return Type

型アノテーションやresponse_modelにより戻り値の型が指定できるという説明。

型アノテーションではAny、response_modelではUserOutクラスを指定した場合、
①POSTのリクエストとしてはUserInオブジェクトが入力される。
②userにUserInオブジェクトが代入される
③return userでuserが返される。
④response_model=UserOutからuserはUserOutに変換される。

.py

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


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

クラスの継承を用いればresponse_modelなしでも同じようなことができるが、レスポンスのデータ量が増えるためresponse_modelを使うことが推奨されている。

下記の例だとresponse_model=Noneがないと失敗する。
response_model=Noneと指定することでFastAPIデフォルトのデータ検証を無効にしている。

.py

from fastapi import FastAPI, Response
from fastapi.responses import RedirectResponse

app = FastAPI()


@app.get("/portal", response_model=None)
async def get_portal(teleport: bool = False) -> Response | dict:
    if teleport:
        return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
    return {"message": "Here's your interdimensional portal."}

response_model_exclude_unset=Trueを指定すればレスポンスに初期値を含まないように設定できる。

.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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

response_model_includeとresponse_model_excludeを使えばレスポンスの項目を制御できる。(nameはレスポンスに含ませるけどpassは除外するなど)

.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
    "baz": {
        "name": "Baz",
        "description": "There goes my baz",
        "price": 50.2,
        "tax": 10.5,
    },
}


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


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

Extra Models

オブジェクトを辞書型に変換、辞書型をもとにオブジェクト生成などの説明。
response_model=Union[PlaneItem, CarItem]やresponse_model=List[Item]やresponse_model=Dict[str, float]を使えばレスポンスが複数ある場合の型指定などができる。

Response Status Code

status_code=status.HTTP_201_CREATEDをなどによりレスポンスのステータスコードを指定できる。

フォームデータ

Queryと同様の使用方法でFormを用いれば、フォームのパラメータもバリデーションなどが可能。
例が少ないため未検証。

.py

from fastapi import FastAPI, Form

app = FastAPI()


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

Request Files

Queryと同様の使用方法でFileを用いれば、ファイルに関するバリデーションなどが可能。
UploadFileを用いれば、filenameによるファイル名の取得やwrite(data)によるファイル書き込みが可能。
また、Listを用いれば複数ファイルのアップロードに対応可能。

.py

from typing import Annotated

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

app = FastAPI()


@app.post("/files/")
async def create_files(
    files: Annotated[list[bytes], File(description="Multiple files as bytes")],
):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(
    files: Annotated[
        list[UploadFile], File(description="Multiple files as UploadFile")
    ],
):
    return {"filenames": [file.filename for file in files]}


@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<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 Forms and Files

Queryと同様の使用方法でFileやFormを用いれば、ファイルやフォームに関するバリデーションなどが可能。

.py

from typing import Annotated

from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()


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

Handling Errors

raise HTTPException()によりHTTPステータスコードを指定した例外処理を発生させることができる。
@app.exception_handler()により例外時の処理を記載して、JSONResponseでレスポンスを作成可能。

下記コードについて
①http://localhost:8000/items/3にアクセス
②HTTPException()が発生
③http_exception_handler()が処理される
または
①http://localhost:8000/items/aを入力
②RequestValidationErrorが発生
③validation_exception_handler()が処理される

main.py

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


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


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

FastAPIのhttp_exception_handlerを再利用することで例外発生時に独自の処理を入れ込むことが可能。

main.py

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(request, exc)

Path Operation Configuration

status用いればHTTP_201_CREATEDなどを用いることができる。

tagsを用いればSwagger UIを下記のような表記ができる。

.py

@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
    return item


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


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

summaryやdescriptionやresponse_descriptionを用いてパスオペレーションの説明が可能。

.py

@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    description="Create an item with all the information, name, description, price, tax and a set of unique tags",
    response_description="The created item",
)
async def create_item(item: Item):
    return item

また下記のようにmarkdown形式でコメントを書くことも可能。

.py

@app.post("/items/", response_model=Item, summary="Create an item")
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

deprecated=Trueとすることで非推奨なパスとしてSwagger UI上に表記できる。

JSON Compatible Encoder

jsonable_encoder()がオブジェクトをJSON型に変換してくれる。

ボディ - 更新

PATCHにより既存の辞書型変数を部分的な更新ができる。

.py

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

Dependencies

下記の例だと2つのAPIからCommonsDep→Depends(common_parameters)と呼び出され依存関係(≒共通処理？)がある。

main.py

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

Security

OAuth2.0などに関する説明。
FastAPIのHTTPExceptionにはヘッダーが存在するため、OAuth 2.0などに対応可能。

ミドルウェア

ミドルウェアを用いたコーディング。
@app.middleware("http")とすればhttpの共通処理が定義できる。

.py

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

CORS (オリジン間リソース共有)

CORSに関する制約を追加することができる。
リソース共有を許可するオリジンを定義したりできる。

.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def main():
    return {"message": "Hello World"}

SQL (Relational) Databases

長くなりそうなので別記事で作成予定。

Bigger Applications - Multiple Files

長くなりそうなので別記事で作成予定。

Background Tasks

BackgroundTasksを用いることで応答後の処理を設定できる。

.py

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

Metadata and Docs URLs

FastAPIのメタデータの使い方が記載されている。
descriptionやtags_metadataやlicense_infoなどの説明。

静的ファイル

こちらの記事のほうがわかりやすい。
https://qiita.com/rubberduck/items/3734057d92a5ee7a2e83

静的なファイルをマウントできる。

下記の2ファイルを用意する。

main.py

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static", html=True), name="static")


@app.get("/static/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

static\index.html

<h1>test</h1>

http://localhost:8000/static/にアクセスすると、
index.htmlの内容が表示される。

テスト

$ pip install pytestしてテスト可能な環境が完成。

from .main import appではなくfrom main import appとしないと動作しないため注意。

これで$ pytestも問題なく動作する。
VScodeなら関数単位でテスト可能。

テスト関数をデバックする場合▶を右クリックして、デバックを選択する。

デバッグ

下記のようにimport uvicornとif __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)が記載してあればデバックできる。

VScodeの場合は下記から「実行とデバック」を選択して、Pythonを選択する。

http://localhost:8000/docsからAPIを実行することでデバックが可能となる。
直接URLを叩いてもデバックできない。

おわり

チュートリアルだけでも幅広く、半分程度しか理解できませんでした。
実際にFastAPIを使うときは必要に応じて「高度なユーザーガイド」も参照する必要がありそうです。
気が向いたらこの記事をまた整理しようと思います。