Ricebox FastAPI

main.py

import re
from typing import Annotated, List, Optional, Dict, Any
from annotated_types import Interval
from sqlmodel import Field, Session, SQLModel, create_engine, select
from pathlib import Path
from fastapi import FastAPI, HTTPException, Depends
from models import Dish, Ricebox, DishAttributes, RiceboxV2, \
    GetDishV2Response, AddDishV2Response, RemoveDishV2Response, GetRiceboxV2Request


# 定義環境及"constant"
CSV_FILE = 'thisthisrice.csv'
DB_PATH = 'rice.db'
SQLDB_PATH = 'sqlite:///' + DB_PATH

# Importing data
if Path(DB_PATH).exists():
    engine = create_engine(SQLDB_PATH)
    print("DB已建立，跳過CSV")
else:
    engine = create_engine(SQLDB_PATH)
    with open(CSV_FILE, 'r', encoding='utf-8') as file_descriptor:
        csv_content: list[str] = file_descriptor.readlines()
    # 移除最後一行
    csv_content.pop()
    pattern = re.compile(r"[0-9\. \n\*]+")
    fields_list: list[str] = []
    for line in csv_content:
        fields_list.append(line.split(","))
    # 問題1：*在這裡的用處；zip的功能
    dish_by_wday = list(zip(*fields_list))

    # 把Dish儲存到SQLite
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        for i in range(len(dish_by_wday)):
            for x in dish_by_wday[i]:
                session.add(Dish(name=pattern.sub('', x), wday=i))
            session.commit()
            

app = FastAPI()

ricebox = RiceboxV2()


@app.get("/get_ricebox")
def get_ricebox() -> GetDishV2Response:
    global ricebox
    response = ricebox.get_ricebox(GetRiceboxV2Request(
        dishes=ricebox.dishes,
        dish_count=ricebox.dish_count,
        isFish=ricebox.isFish,
        price=ricebox.price
    ))
    return response


@app.post("/add")
def add(dish: Dict) -> AddDishV2Response:
    global ricebox
    response = ricebox.add(Dish(name=dish["name"], wday=dish["wday"]))
    return response


@app.post("/remove")
def remove(dish: Dict) -> RemoveDishV2Response:
    global ricebox
    response = ricebox.remove(dish_name=dish["name"])
    return response
#=========================================================================#

models.py

from pydantic import BaseModel, computed_field, ValidationError
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Annotated, List, Optional, Dict
from annotated_types import Interval
from enum import Enum
from fastapi import HTTPException


class DishAttributes(Enum):
    FISH_SET_KEYWORD: str = '蒸倉魚'
    MAX_NO_OF_DISHES: int = 4
    MAX_NO_OF_FISH_DISHES: int = 3
    MIN_NO_OF_DISHES: int = 2
    BASE_PRICE: int = 35
    PERDISH_PRICE: int = 8
    FISH_BASE_PRICE: int = 58


class Dish(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    wday: Annotated[int, Interval(ge=0, lt=6)]

    @computed_field
    def is_meat(self) -> bool:
        _meat_keywords: List[str] = ['牛', '豬', '雞', '魚', '羊', '叉燒', '肉', '班腩', '鴨', '鵝']
        for m in _meat_keywords:
            if m in self.name:
                return True
        return False
      
      
class AddDishV2Response(BaseModel):
    dishes: Optional[List[str]]
    dish_count: int
    isFish: bool
    price: int


class RemoveDishV2Response(BaseModel):
    dishes: Optional[List[str]]
    dish_count: int
    isFish: bool
    price: int


class GetDishV2Response(BaseModel):
    dishes: Optional[List[str]]
    dish_count: int
    isFish: bool
    price: int


class GetRiceboxV2Request(BaseModel):
    dishes: Optional[List[str]]
    dish_count: int
    isFish: bool
    price: int
      
      
class RiceboxV2:
    isFish: bool = False
    dish_count: int = 0
    dishes: Optional[List[Dish]] = Field(default=None)

    def __init__(self) -> None:
        self.price: int = 0
        self.dish_count: int = 0
        self.dishes = []

    def add(self, dish: Dish) -> AddDishV2Response:
        if DishAttributes.FISH_SET_KEYWORD.value in dish.name:
            self.isFish = True
            self.dish_count += 1
            if self.dish_count > int(DishAttributes.MAX_NO_OF_FISH_DISHES.value):
                raise ValueError(f"你個魚飯唔可以多個{DishAttributes.MAX_NO_OF_FISH_DISHES.value}個餸！")
            self.price = int(DishAttributes.FISH_BASE_PRICE.value)
        else:
            self.dish_count += 1
            if self.dish_count > int(DishAttributes.MAX_NO_OF_DISHES.value):
                raise ValueError(f"你個飯唔可以多個{DishAttributes.MAX_NO_OF_DISHES.value}個餸！")
            self.price = int(DishAttributes.BASE_PRICE.value)

        if self.dish_count > 2:
            self.price += int(DishAttributes.PERDISH_PRICE.value)

        self.dishes.append(dish.name)
        return AddDishV2Response(
            dishes=self.dishes,
            dish_count=self.dish_count,
            isFish=self.isFish,
            price=self.price
        )

    def remove(self, dish_name: str) -> RemoveDishV2Response:
        if dish_name in self.dishes:
            self.dishes.remove(dish_name)
            self.dish_count -= 1

            is_fish_set_dish_present = any(DishAttributes.FISH_SET_KEYWORD.value in item for item in self.dishes)

            if self.dish_count == 0:
                self.price = 0
            elif is_fish_set_dish_present:
                self.price = int(DishAttributes.FISH_BASE_PRICE.value)
            else:
                if self.dish_count >= 2:
                    self.price -= int(DishAttributes.PERDISH_PRICE.value)
                else:
                    self.price = int(DishAttributes.BASE_PRICE.value)

            return RemoveDishV2Response(
                dishes=self.dishes,
                dish_count=self.dish_count,
                isFish=self.isFish,
                price=self.price
            )
        else:
            raise HTTPException(status_code=404, detail="Dish not found")

    def get_ricebox(self, ricebox: GetRiceboxV2Request) -> GetDishV2Response:
        return GetDishV2Response(
            dishes=ricebox.dishes,
            dish_count=ricebox.dish_count,
            isFish=ricebox.isFish,
            price=ricebox.price
        )

requirements.txt

fastapi~=0.110.1
httptools
pip-chill
python-dotenv
pyyaml
sqlmodel~=0.0.16
uvicorn
uvloop
watchfiles
websockets

pydantic~=2.6.4
SQLAlchemy~=2.0.29