FastAPI モジュール - APIRouter

FastAPI モジュール - 基本 - Qiita
FastAPI モジュール - APIRouter - Qiita

前回、FastAPIで大規模なアプリを構築するためのモジュール分割について説明しました。
前回はusersモジュールだけでしたが、今回はitemsモジュールを追加したいと思います。

またこの記事全体のソースは以下のドキュメントであり、ここでは部分的な説明を順序をつけて加えています。
Bigger Applications - Multiple Files - 公式サイト

１．itemesモジュール

１－１．ディレクトリ構成

items.pyを追加します

.
├── app                  # "app" はPython パッケージ
│   ├── __init__.py      # "app" を "Python パッケージ"にする
│   ├── main.py          # "main" モジュール, e.g. import app.main
│   ├── dependencies.py  # "dependencies" モジュール, e.g. import app.dependencies
│   └── routers          # "routers" は"Python サブパッケージ"
│   │   ├── __init__.py  # "routers" を"Python サブパッケージ"にする
│   │   ├── items.py     # "items" サブモジュール, e.g. import app.routers.items
│   │   └── users.py     # "users" サブモジュール, e.g. import app.routers.users

１－２．items.py

app/routers/items.py

from fastapi import APIRouter, Depends, HTTPException
from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}

@router.get("/")
async def read_items():
    return fake_items_db

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

@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

ここでは以下のAPIRouterクラスに注目してください。items.pyの3つの path operations 全部に共通の指定を行っています。

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

prefix="/items"　ですから "/" は "/items/"の意味になります。"/{item_id}" は "/items/{item_id}" を指します。
tags=["items"]　は全てのpath operationsに同じ指定を行っていることになります。dependencies=[Depends(get_token_header)] も 全path operation decoratorsで指定されている意味です。responsesも同様に全パスの仕様に追加されます。

dependencyについては、mainでGlobal Dependenciesを指定しているので、通算で２個指定していることになります。

また以下のpath operationsでは個別にtagsを指定しているので、SwaggerUI上はこのパスは、"items"と"custom"の、２回現れます。

@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)

FastAPI モジュール - 基本

１－３．dependencies.py

今回はget_token_header関数を追加しています。上のitemsの説明でみたように、itemsの全path operationsのdependenciesとして有効です。この関数はX-Token headeを読み取るdependencyです。

app/dependencies.py

from fastapi import Header, HTTPException

async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

１－４．main.py

include_router()でitems.routerを読み込み、appに追加しています。

app/main.py

from fastapi import Depends, FastAPI
from .dependencies import get_query_token
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])

app.include_router(users.router)
app.include_router(items.router)

@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

Dependencyについては以下の公式サイトに詳しい説明があります。
Dependencies in path operation decorators - 公式サイト

２．SwaggerUI

２－１．初期画面

今回はitemsとcusutomのタグが追加されています。

２－２．２つのtoken

リクエストの、要求されているパラメータを確認すると、2つのtokenがあります。これはmainのGlobal Dependencies(token)と上の１－３．のDependencies(x-token)に対応するものです。

２－３．実行

以下のパラメータ値を入力して実行します。
token=jessica , x-token=fake-super-secret-token

うまく動作しているようです。まあ本記事の目的は、詳細よりは全体の流れなので、エラー値とかは追いません。

今回は以上です。