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"
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 = info.context
print(context)
if v != "ins01":
raise ValueError()
return v
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)
):
users = db.query(User).all()
names = [user.name for user in users]
action.validate(request.query_params, context={"names": names})
return action
解説
CreateInstance アクションの場合は create_instance のルーティングを使います
@app.get などは付与せずにただの非同期関数として定義します
もし create_instance をインタフェースとして公開したい場合には @app.get などを付与して公開することができるようにしておきます
root ルーティングでは Action 名に応じてルーティングを決定しています
バリデーション時も同じような判定をしているのでここはもう少しうまくやればキレイに書けるかもしれません
最後に
Action 名ごとにルーティングを定義する場合には Action 名ごとにルーティングを作成するとコードがキレイになるかもしれません
