Site Overlay

簡単なWebサービスをPythonのみで作ってみた２

投稿者: Kenji

前回に引き続きStreamlit/FastAPI/MongoDBで作ったウェブアプリについてアップしていきます。

4)実装コード　FastAPI側

FastAPのインストールなどについては、例えば こちらの1)背景で紹介したページなどを参照してください。「FastAPI インストール」などでググればたくさん出てきますので、ここでは割愛します。

FastAPIは以下で立ち上がります。

uvicorn main:app --reload

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")

まず、「python main.py」でFastAPIを実行できるように

if __name__ == "__main__":
    uvicorn.run("main:app", host="localhost", port=8000, reload=True)

を追加しています。これが必要な理由はVSCodeなどで、デバッグできるようにするためとなります。

次に、

user_repo: UserRepository = UserRepository()
user_service: UserService = UserService(user_repo)

ですが、変数user_repositoryは永続化に関するインスタンスで、user_serviceがビジネスロジックに関するインスタンスです。

httpメソッドに則ったメソッドを準備します。FastAPIがhttpメソッドに則ったメソッドであることを認識させるために、各メソッドの直上にデコレータでhttpメソッドとURIを記述します：

@app.get("/api/v2/users/")
async def get_users() -> list:
....

@app.post("/api/v2/users/")
async def create_user(payload: str = Body()) -> str:
....
etc.

FastAPIでリクエストのデータを取得する方法の１つに、Body()があります。今回Streamlit側で設定したpayloadからデータを取得するため、「payload: str = Body()」と実装しています。

さて、引き続きuser_service.pyの実装を見てみます。コードは以下の通りです：

from models import *
from users_repository import UserRepository


class UserService:
    user_repo: UserRepository

    def __init__(self, user_repo: UserRepository) -> None:
        self.user_repo = user_repo

    def find(self, sequence_nbr: int) -> (User | None):
        result = self.user_repo.get_user(sequence_nbr)
        return result

    def find_all(self) -> list:
        return self.user_repo.get_all_users()

    def register(self, sequence_nbr: int, first_name: str, last_name: str, gender: str, roles: str) -> bool:
        user = User(sequence_nbr = sequence_nbr, first_name = first_name, 
                    last_name = last_name, gender = gender, roles = roles)
        result = self.user_repo.create_user(user)
        return result

    def update(self, sequence_nbr: int, first_name: str, last_name: str) -> bool:
        command = UpdateUserCommand(sequence_nbr = sequence_nbr, first_name = first_name, last_name = last_name)
        result = self.user_repo.update_user(command)
        return result

    def remove(self, sequence_nbr: int) -> bool:
        result = self.user_repo.delete_user(sequence_nbr)
        return result

    def remove_all(self) -> bool:
        result = self.user_repo.delete_all()
        return result

今回のアプリはほとんどビジネスロジックがないので、シンプルなものになっています。

最後に永続化周りの「user_repository.py」です：

from typing import List
from pymongo import MongoClient
from models import *


class UserRepository():

    DB_URL = 'mongodb://localhost:27017'

    def __init__(self) -> None:
        self.client: MongoClient = MongoClient(__class__.DB_URL)
        self.db = self.client.users
        self.collection = self.db.users

    def get_all_users(self) -> list:
        found_users = []
        cursor = self.collection.find()
        for document in cursor:
            found_users.append(User(**document))
        return found_users

    def get_user(self, sequence_nbr: int) -> (User | None):
        found_user = self.collection.find_one({"sequence_nbr": sequence_nbr})
        if found_user:
            return User(**found_user)

    def create_user(self, user: User) -> bool:
        insertOneResult = self.collection.insert_one(user.dict())
        return insertOneResult.acknowledged

    def create_many_users(self, user_list: List[User]) -> bool:
        insertManyResult = self.collection.insert_many(user_list)
        return insertManyResult.acknowledged

    def update_user(self, command: UpdateUserCommand) -> bool:
        updateResult = self.collection.update_one({"sequence_nbr": command.sequence_nbr},
                                                  {"$set": {"first_name": command.first_name,
                                                            "last_name": command.last_name}})
        return updateResult.acknowledged

    def delete_user(self, sequence_nbr: int) -> bool:
        deleteResult = self.collection.delete_one(
            {"sequence_nbr": sequence_nbr})
        return deleteResult.acknowledged

    def delete_all(self) -> bool:
        deleteResult = self.collection.delete_many({})
        return deleteResult.acknowledged

DB操作周りの実装なるべく１つに閉じさせたいため、user_repository.pyでMongoDBを操作する実装を記述しています。

    def __init__(self) -> None:
        self.client: MongoClient = MongoClient(__class__.DB_URL)
        self.db = self.client.users
        self.collection = self.db.users

pythonでMongoDBを操作するには、「pymongo」というライブラリが必要です。インストールは「pip install pymongo」でできます。詳細については前回の記事の1-3)で紹介したリンクを参照ください。MongoDBでは、
　db：データベース名
　collection：テーブル名
を表すイメージとなります。一旦今回はデータベース名もテーブル名も「users」にしています。

あとはCRUDに応じたメソッドを実装すればよいのですが、これは上記の通りです。詳細はpymongoの公式ドキュメントに譲りますが、ざっくりは以下の通りです。下記以外にもいろいろメソッドが提供されているようです。

Createinsert_one()：１レコードのみ作成
insert_many()：複数レコード作成
Readfind_one()：1レコードのみ取得
find()：全レコードを取得
Updateupdate_one()：１レコード更新
Deletedelete_one()：1レコードのみ削除
delete_many()：全レコード削除

※まぁ、メソッド名から想像できる通りの内容です。

—–20230416追記 開始——
※models.pyを記載していなかったので、追加です。

from typing import List
from pydantic import BaseModel
from enum import Enum
from pydantic import BaseModel


class Gender(str, Enum):
    male = "male"
    female = "female"


class Role(str, Enum):
    admin = "admin"
    user = "user"


class User(BaseModel):
    sequence_nbr: int
    first_name: str
    last_name: str
    gender: Gender
    roles: Role


class UpdateUserCommand:
    sequence_nbr: int
    first_name: str
    last_name: str

    def __init__(self, sequence_nbr: int, first_name: str, last_name: str) -> None:
        self.sequence_nbr = sequence_nbr
        self.first_name = first_name
        self.last_name = last_name

今回シンプルなアプリなので、modelもシンプルな形になってます。
—–20230416追記 終了——

以上がバックエンドの実装内容です。バック側もFastAPI/MongoDBを使うとシンプルに書けることがわかるかと思います。

次の記事で、もう少し踏み込んで、ドメイン駆動設計(DDD)の考え方を取り入れたコードにしたいと思います。DDDがエリック•エバンズによって提唱されたのは20年ほど前のことですが、最近トレンドとなっているマイクロサービスと相性がいいので、注目されている設計手法です。

今回は以上です。

最後まで読んでいただきありがとうございます。
質問等はコメント欄かお問合せにてよろしくおねがいいたします。

「簡単なWebサービスをPythonのみで作ってみた２」への1件のフィードバック

  1. ピンバック: 簡単なWebサービスをPythonのみで作ってみた３ - 一般サラリーマンによる数学・理科を語るブログ

コメントを残す

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です