こんにちは、本日はFastAPIのサブクラス化をしてみます。
FastAPIを使ってある程度のWebサービスを構築する際、main.pyにずらずらメソッドを追加していくのは個人的にあまり好きではないので、クラス化します。いろいろネットで調べてみたのですが、意外とサブクラスにする記事がなかったので、勝手に本稿で記載してみます。
こちらで作成したFastAPIのmain.pyを再掲します。
import json
from fastapi import FastAPI, HTTPException, Body
import uvicorn
from users_repository import UserRepository
from user_service import UserService
from models import *
app: FastAPI = FastAPI()
user_repo: UserRepository = UserRepository()
user_service: UserService = UserService(user_repo)
if __name__ == "__main__":
uvicorn.run("main:app", host="localhost", port=8000, reload=True)
@app.get("/")
async def root():
return "うぃ〜〜"
@app.get("/api/v2/users/")
async def get_users() -> list:
return user_service.find_all()
@app.get("/api/v2/users/{sequence_nbr}")
async def get_user_by_seqnbr(sequence_nbr: int) -> User:
result = user_service.find(sequence_nbr)
if result:
return result
raise HTTPException(status_code=404, detail=f"sequence_nbr : {sequence_nbr} not found.")
@app.post("/api/v2/users/")
async def create_user(payload: str = Body()) -> str:
data = json.loads(payload)
result = user_service.register(
int(data["sequence_nbr"]), data["first_name"], data["last_name"], data["gender"], data["roles"]
)
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"user.sequence_nbr : {payload.sequence_nbr} Failed.")
@app.put("/api/v2/users/{sequence_nbr}")
async def update_user(payload: str = Body()) -> str:
data = json.loads(payload)
result = user_service.update(int(data["sequence_nbr"]), data["first_name"], data["last_name"])
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"sequence_nbr = { payload.sequence_nbr } not found")
@app.delete("/api/v2/users/{sequence_nbr}")
async def delete_user(sequence_nbr: int) -> str:
result = user_service.remove(sequence_nbr)
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"Delete user failed, sequence_nbr = {sequence_nbr} not found.")
@app.delete("/api/v2/users/")
async def delete_all() -> str:
result = user_service.remove_all()
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"Delete user_list failed")
ここまでメソッドがずらずらあると、なんかクラス化したくなるかと思います。というわけでFastAPIを継承してサブクラスを作成し、main.pyから分離してみます。
もう結果を記載すると以下の通りです。まずmain.pyからです。
from fastapi import FastAPI
import uvicorn
from user_api import UserRegistrationAPI
from models import *
app: FastAPI = UserRegistrationAPI()
if __name__ == "__main__":
uvicorn.run("main:app", host="localhost", port=8000, reload=True)だいぶスッキリしました。次にuser_api.pyです。
import json
from fastapi import FastAPI, HTTPException, Body
from users_repository import UserRepository
from user_service import UserService
from models import *
class UserRegistrationAPI(FastAPI):
user_repo: UserRepository
user_service: UserService
def __init__(self) -> None:
super().__init__()
self.user_repo = UserRepository()
self.user_service = UserService(self.user_repo)
self.link_routes()
def link_routes(self):
routes = self.get_routes()
for route in routes:
path, handler, methods = route
self.add_api_route(path, handler, methods=methods)
def get_routes(self):
return [
("/", self.root , ["GET"]),
("/api/v2/users/", self.get_users , ["GET"]),
("/api/v2/users/{sequence_nbr}", self.get_user_by_seqnbr , ["GET"]),
("/api/v2/users/", self.create_user , ["POST"]),
("/api/v2/users/{sequence_nbr}", self.update_user , ["PUT"]),
("/api/v2/users/{sequence_nbr}", self.delete_user , ["DELETE"]),
("/api/v2/users/", self.delete_all , ["DELETE"]),
]
async def root(self):
return "うぃ〜〜"
async def get_users(self) -> list:
return self.user_service.find_all()
async def get_user_by_seqnbr(self, sequence_nbr: int) -> User:
result = self.user_service.find(sequence_nbr)
if result:
return result
raise HTTPException(status_code=404, detail=f"sequence_nbr : {sequence_nbr} not found.")
async def create_user(self, payload: str = Body()) -> str:
data = json.loads(payload)
result = self.user_service.register(
int(data["sequence_nbr"]), data["first_name"], data["last_name"], data["gender"], data["roles"]
)
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"user.sequence_nbr : {payload.sequence_nbr} Failed.")
async def update_user(self, payload: str = Body()) -> str:
data = json.loads(payload)
result = self.user_service.update(int(data["sequence_nbr"]), data["first_name"], data["last_name"])
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"sequence_nbr = { payload.sequence_nbr } not found")
async def delete_user(self,sequence_nbr: int) -> str:
result = self.user_service.remove(sequence_nbr)
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"Delete user failed, sequence_nbr = {sequence_nbr} not found.")
async def delete_all(self) -> str:
result = self.user_service.remove_all()
if result:
return "Success!!"
raise HTTPException(status_code=404, detail=f"Delete user_list failed")
修正前のmain.pyで「@app.get(“/”)」などとしていたデコレータは、そのままではUserRegistrationAPIクラスにもってこれないので、
FastAPIクラスのメソッド「add_api_route」を用いて、UserRegistrationAPIの各メソッドとエンドポイントを紐づけています。
その紐付けを「link_routes」メソッドにまとめ、コンストラクタ__init__で実施している形です。
本日は短いですが、以上です。かなりニッチですが、本記事がだれかの役に立てば幸いです。
最後まで読んでいただきありがとうございます。
質問等はコメント欄かお問合せにてよろしくおねがいいたします。