fastapi で AWS のような Action 名ベースの API をどう構築するか考える (ルーティング編)

• 概要
• 環境
• 方針
• サンプルコード
• 解説
• 最後に

概要

前回バリデーションをどうするか検討しました
今回はさらにルーティングをどう管理するか考えます

環境

• Python 3.11.3
• fastapi 0.100.1
• pypdantic 2.1.1

方針

• ユーザインタフェースとなるルーティングは一つだが内部では Action ごとにルーティングを分けたい
• root で一旦すべてのルーティングを受けて Action 名で判断し適切なルーティングをコールする

サンプルコード

from enum import Enum

from fastapi import Depends, FastAPI, Query, Request
from fastapi.datastructures import QueryParams
from pydantic import BaseModel, Field, FieldValidationInfo, field_validator
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

app = FastAPI()


engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


def get_db():
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()


# 各種アクション名を管理する列挙型のクラス
class ActionName(Enum):
    CREATE_INSTANCE = "CreateInstance"


# CreateInstance(例) という名前のアクションのバリデーションを管理するモデル
class CreateInstance(BaseModel):
    instance_name: str = Field(alias="InstanceName")

    @field_validator("instance_name")
    @classmethod
    def validate_instance_name(cls, v: str, info: FieldValidationInfo) -> str:
        # context は FieldValidationInfo に格納されているので使用する場合は引数に追加する
        context = info.context
        # データベースから取得した値が取得できていることが確認できる
        print(context)
        if v != "ins01":
            raise ValueError()
        return v


# 起点となる Action 名のバリデーションを管理するモデル
class Action(BaseModel):
    action: str = Field(Query(alias="Action"))

    @field_validator("action")
    @classmethod
    def validate_action(cls, v: str) -> str:
        # ここでは許可するバリデーション名など基本的なバリデーションのみを行う
        if v not in [ActionName.CREATE_INSTANCE.value]:
            raise ValueError()
        return v

    # アクション名に応じて各種バリデーションをコールする
    def validate(self, query_params: QueryParams, context: dict):
        if self.action == ActionName.CREATE_INSTANCE.value:
            CreateInstance.model_validate(query_params, context=context)


@app.get("/")
async def root(
    request: Request, action: Action = Depends(), db: Session = Depends(get_db)
):
    if action == ActionName.CREATE_INSTANCE.value:
        return await create_instance(request, action, db)


async def create_instance(
    request: Request, action: Action = Depends(), db: Session = Depends(get_db)
):
    # データベースの値はここで取得してバリデーション対応の値を context として渡す
    users = db.query(User).all()
    names = [user.name for user in users]
    # ルーティングの引数で Request オブジェクトを参照できるのでこれを使って各種アクションのバリデーションをコールする
    action.validate(request.query_params, context={"names": names})
    return action

解説

CreateInstance アクションの場合は create_instance のルーティングを使います
@app.get などは付与せずにただの非同期関数として定義します

もし create_instance をインタフェースとして公開したい場合には @app.get などを付与して公開することができるようにしておきます

root ルーティングでは Action 名に応じてルーティングを決定しています
バリデーション時も同じような判定をしているのでここはもう少しうまくやればキレイに書けるかもしれません

最後に

Action 名ごとにルーティングを定義する場合には Action 名ごとにルーティングを作成するとコードがキレイになるかもしれません