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

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__()

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)

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)

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

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,
        }