Webアプリ開発初心者による「豆ログ」の開発１ ~FastAPIでバックエンド開発~

gotoooo 2ヶ月前

構想

過去にGlideで作った自分用のコーヒー豆購入履歴管理ツールをWebアプリ化してみたい。

gotoooo 2ヶ月前

構想-アーキテクチャ

流行ってそうなフレームワークを使ってみる。

フロントエンド：React
　選定理由：2024年3月時点で新規開発ならReact+TypeScript一択な風潮があるため

バックエンド：FastAPI
　選定理由：フロントエンドの学習で手一杯になることが予想され、使用経験のあるPythonで言語学習コストを抑えるため

DB: Mircrosoft SQL Server
　選定理由：実務で使用経験があるため

gotoooo 2ヶ月前

流れ

API仕様検討
バックエンド開発＋DBスタブ実装
フロントエンド開発
DB準備＋バックエンド改修
どこかにデプロイ

作るものの仕様が決まっているので機能設計、画面設計などは割愛。

gotoooo 2ヶ月前

FastAPIはコードファーストでAPIのドキュメントが生成される。
なので1.API仕様検討と2.バックエンド開発　は並行して進める。

gotoooo 2ヶ月前

FastAPI開発準備として非常に参考になる

gotoooo 2ヶ月前に更新

2.バックエンド開発＋DBスタブ実装

準備

リポジトリ作成
Githubにて作成。バックエンドとフロントエンドは分けた。
Python仮想環境作成

python -m venv .venv

仮想環境名を何にするか小一時間悩んでしまった。
元々.envとしていたが、これは環境変数を収めるファイルとして使用されることが多いらしく、コンフリクトしてしまう。ひとまず.venvとすることにした。

ライブラリインストール

fastapi
バックエンドフレームワーク
uvicorn
ASGI準拠のPythonアプリを実行するサーバ
ローカル環境のデバッグ時には「uvicorn main:app --reload」でアプリを起動する
gunicorn
WSGI準拠のPythonアプリを実行するサーバ
python-decouple
環境変数を扱う
passlib
ハッシュ値の作成/取得/検証

フォルダ構成作成
こんな感じにした。多分実装が進むにつれて変わる。

api
アプリ動作制御
databases
対向するDBに対する処理を実装する。
テストや動作確認を容易にするため、DBのスタブもここに実装する。
routers
APIのルーティングを実装する
schemas
APIのリクエスト、レスポンスのデータ定義を実装する。

メモ：DDDでいうところのEntityやDBに出し入れするデータ構造定義と似て非なるレイヤーである。
ここに実装したクラスを使いまわしても良いものか、別途定義してデータの詰替を実装すべきか悩ましい。
=> ChatGPTに聞いたところ、APIのIn/OutとDBのIn/Outとで同じクラスを使い回すのは推奨されているが、これらのデータ定義に差異がある場合は別クラスにすべき　とのこと。
　なるほど、未来永劫同じデータ構造とも限らないので別レイヤーでそれぞれデータ構造を定義することとする。クラス名が同じだと紛らわしいのでDBに収める方のはxxxDtoとする。

services

gotoooo 2ヶ月前に更新

DDDの概念を取り入れると同じデータモデルを最大で３つのクラスで扱うことになる。

APIのIn/Out
DBのIn/Out
ドメインモデル

DB <--> (リポジトリ) <--> ドメインモデル <--> API

ドメインモデルのレイヤーで各データ構造に変換する処理を実装しておけば良さそう。

gotoooo 2ヶ月前

アクセス先のDBを本物とスタブとで切り替えるためのリポジトリのI/F的なクラスを作成した。

コード例

from abc import ABC, abstractmethod
from typing import List, Generic, TypeVar
from .entities import UserModel, RollModel, ProductionAreaModel, StoreModel, BrandModel, PurchaseHistoryModel, ImpressionModel

TEntity = TypeVar('TEntity') 

class RepositoryBase(ABC, Generic[TEntity]):
    def __init__(self) -> None:
        super().__init__()

    @abstractmethod
    def add(self, entity: TEntity) -> TEntity:
        pass

    @abstractmethod
    def get(self, id: str) -> TEntity:
        pass

    @abstractmethod
    def get_all(self) -> List[TEntity]:
        pass

    @abstractmethod
    def update(self, id: str, entity: TEntity) -> None:
        pass

    @abstractmethod
    def delete(self, id: str) -> None:
        pass

class UserRepository(RepositoryBase[UserModel]):
    def __init__(self) -> None:
        super().__init__()

class RollRepository(RepositoryBase[RollModel]):
    def __init__(self) -> None:
        super().__init__()

class ProductionAreaRepository(RepositoryBase[ProductionAreaModel]):
    def __init__(self) -> None:
        super().__init__()

class StoreRepository(RepositoryBase[StoreModel]):
    def __init__(self) -> None:
        super().__init__()

class BrandRepository(RepositoryBase[BrandModel]):
    def __init__(self) -> None:
        super().__init__()

class PurchaseHistoryRepository(RepositoryBase[PurchaseHistoryModel]):
    def __init__(self) -> None:
        super().__init__()

class ImpressionRepository(RepositoryBase[ImpressionModel]):
    def __init__(self) -> None:
        super().__init__()

gotoooo 2ヶ月前に更新

DBのスタブへのアクセス部分を実装した。
リポジトリのコンストラクタで適当なエンティティを作成してself.itemsに収めるようにした。
この仕組みによりフロントエンドの動作確認が容易になることを期待している。

コード例

from typing import List, TypeVar

from ..domains.repositories import *
from ..domains.entities import *

TEntity = TypeVar("TEntity", bound=EntityBase)


class FakeRepositoryBase(RepositoryBase[TEntity]):
    def __init__(self) -> None:
        super().__init__()
        self.items: List[TEntity] = []

    def add(self, entity: TEntity) -> TEntity:
        uid = f"{len(self.items)}"
        entity.id = uid
        self.items.append(entity)

        return entity

    def get(self, id: str) -> TEntity:
        return next((item for item in self.items if item.id == id), None)

    def get_all(self) -> List[TEntity]:
        return self.items

    def update(self, id: str, entity: TEntity) -> None:
        to_update = self.get(id)
        if to_update == None:
            return
        i = self.items.index(to_update)

        # TODO: 論理削除とするか物理削除とするかアーカイブとするか
        # 更新の場合は一旦レコードを削除して同じIDで追加で問題なさそう
        self.items.remove(to_update)

        entity.updated_at = datetime.now()
        self.items.insert(i, entity)

    def delete(self, id: str) -> None:
        to_update = self.get(id)
        if to_update == None:
            return
        i = self.items.index(to_update)

        # TODO: 論理削除とするか物理削除とするかアーカイブとするか
        # 銘柄レコードを削除するとそれを参照する購入履歴レコードが不整合を起こす
        # 削除フラグ立てるだけにしておく
        # self.items.remove(to_update)
        to_update.deleted_at = datetime.now()


class FakeUserRepository(FakeRepositoryBase[UserModel]):
    def __init__(self) -> None:
        super().__init__()
        admin = UserModel()
        admin.id = "0"
        admin.name = "admin"
        admin.login_id = "admin"
        admin.password = "01234567"
        admin.roll_id = "0"
        self.items.append(admin)

gotoooo 2ヶ月前

2.バックエンド開発＋DBスタブ実装

実装

データ定義
最終的にどうなるかわからないが下記3層分のデータクラスを作成した

DBのI/O
ドメイン層のエンティティ
APIのI/O

ORMの活用でDBのI/O用のは統合できるかもしれない。
APIのI/OとDBのI/Oとでデータ構造に差異が生じた場合に備えて最低でも2層分は用意しておく。

リポジトリのI/F定義
CRUDに相当する抽象メソッドをもつ抽象クラスを作成。
DBスタブのリポジトリ実装
エンティティを保持する配列を用意しておき、CRUDメソッドに応じて配列の中身を更新する。
エンドポイント実装
FastAPIのDependsを活用して実行環境に応じてDIするリポジトリを切り替えられるようにする。

使用するリポジトリ切り替えの実装イメージ

from .libs.domains.repositories import UserRepository
from .libs.databases.in_memory import FakeUserRepository

from fastapi import FastAPI, Depends

app = FastAPI()


def get_user_repository() -> UserRepository:
    # DebugやUnitTest時
    r = FakeUserRepository()

    # TODO: 本番環境
    # r = DBUserRepository()
    return r


@app.get("/")
def root(repository: UserRepository = Depends(get_user_repository)):
    users = repository.get_all()
    return {"message": "Hello World!"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("api.main:app", host="0.0.0.0", port=8000, reload=True)

gotoooo 2ヶ月前に更新

DBのスタブor本物の切り替えは専用のFactoryクラスを用意してみた。
.envにUSE_DB_STUB=True/Falseの環境変数を持たせておいてdecoupleでsetting.pyに格納、それを参照する形とした。
多分DI用のライブラリを使うとこの辺のコード量は減らせそうではある。後にリファクタリングする。

コード例

from ...setting import *
from ..domains.repositories import *
from ..utils import Singleton


class Factory(Singleton):
    def __init__(self) -> None:
        self.user_repository = self.init_user_repository()
        self.roll_repository = self.init_roll_repository()
        self.production_area_repository = self.init_production_area_repository()
        self.store_repository = self.init_store_repository()
        self.brand_repository = self.init_brand_repository()
        self.purchase_history_repository = self.init_purchase_history_repository()
        self.impression_repository = self.init_impression_repository()

    def init_user_repository(self) -> UserRepository:
        if USE_DB_STUB == True:
            from ..databases.in_memory import FakeUserRepository

            return FakeUserRepository()
        else:
            # TODO: 本番環境
            return FakeUserRepository()

    def get_user_repository(self) -> UserRepository:
        return self.user_repository

gotoooo 2ヶ月前

エンドポイントの実装を進めた。
フロントエンドから渡ってくるデータと返すデータとで微妙にデータ構造が異なるケースがあるため、schema.pyでデータ定義を分けた。

フロントエンドから渡ってくるデータ：xxxBody
返すデータ：xxxInfo

あとドメイン層のentities.pyで各データモデルにxxxBody, xxxInfoの相互変換メソッドを追加した。

エンドポイントの一例

from fastapi import APIRouter, Request, Response, HTTPException, Depends
from typing import List
from .repository_factory import Factory
from ..domains.repositories import PurchaseHistoryRepository
from ..domains.entities import PurchaseHistoryModel
from ..schemas import SuccessMessage, PurchaseHistoryBody, PurchaseHistoryInfo

router = APIRouter()

fac = Factory.get_instance()


@router.post("/api/purchase_history", response_model=PurchaseHistoryInfo)
async def create_purchase_history(
    request: Request,
    response: Response,
    data: PurchaseHistoryBody,
    repository: PurchaseHistoryRepository = Depends(fac.get_purchase_history_repository),
):
    entity = PurchaseHistoryModel(data)
    created = repository.add(entity)
    return created.to_schema_model()


@router.get("/api/purchase_history", response_model=List[PurchaseHistoryInfo])
async def get_purchase_historys(
    repository: PurchaseHistoryRepository = Depends(fac.get_purchase_history_repository),
):
    entities = repository.get_all()
    if not entities or len(entities) <= 0:
        raise HTTPException(status_code=401, detail="The entity was not found.")

    return [entity.to_schema_model() for entity in entities]


@router.get("/api/purchase_history/{id}", response_model=PurchaseHistoryInfo)
async def get_purchase_history(
    id: str,
    repository: PurchaseHistoryRepository = Depends(fac.get_purchase_history_repository),
):
    entity = repository.get(id)
    if not entity:
        raise HTTPException(status_code=401, detail="The entity was not found.")
    return entity.to_schema_model()


@router.put("/api/purchase_history/{id}", response_model=PurchaseHistoryInfo)
async def update_purchase_history(
    request: Request,
    response: Response,
    id: str,
    data: PurchaseHistoryBody,
    repository: PurchaseHistoryRepository = Depends(fac.get_purchase_history_repository),
):
    entity = PurchaseHistoryModel(data, id)
    updated = repository.update(id, entity)
    if not updated:
        raise HTTPException(status_code=404, detail=f"Update task failed")
    return updated.to_schema_model()


@router.delete("/api/purchase_history/{id}", response_model=SuccessMessage)
async def delete_purchase_history(
    request: Request,
    response: Response,
    id: str,
    repository: PurchaseHistoryRepository = Depends(fac.get_purchase_history_repository),
):
    res = repository.delete(id)
    if res == False:
        raise HTTPException(status_code=404, detail=f"Delete task failed")
    return {"message": "Successfully deleted"}

データ型の変換例

class PurchaseHistoryModel(EntityBase):
    """
    Purchasing history of a coffee brand .
    """

    user_id: str
    brand_id: str
    store_id: str
    purchase_at: datetime
    annotation: Optional[str]

    def __init__(
        self, schemaModel: Optional[PurchaseHistoryBody] = None, id: Optional[str] = None
    ) -> None:
        super().__init__()
        self.user_id = None
        self.brand_id = None
        self.store_id = None
        self.purchase_at = None
        self.annotation = None

        if schemaModel:
            self.user_id = schemaModel.user_id
            self.brand_id = schemaModel.brand_id
            self.store_id = schemaModel.store_id
            self.purchase_at = schemaModel.purchase_at
            self.annotation = schemaModel.annotation

        if id:
            self.id = id

    def to_schema_model(self) -> dict:
        return {
            "id": self.id,
            "user_id": self.user_id,
            "brand_id": self.brand_id,
            "store_id": self.store_id,
            "purchase_at": self.purchase_at,
            "annotation": self.annotation,
        }

gotoooo 2ヶ月前

APIの動作確認はpostmanを使う。

gotoooo 2ヶ月前

長くなるので一旦バックエンドまででCloseする

このスクラップは2ヶ月前にクローズされました