Merge pull request #19 from l7wei/main

refactor: 改善程式碼可讀性

main.py

@@ -20,7 +20,7 @@
             host="0.0.0.0",
             port=int(os.getenv("PORT") or 5000),
             log_level="debug",
-            reload=True,  #  reload the server every time code changes
+            reload=True,  # reload the server every time code changes
         )
     else:
         # Production

src/__init__.py

@@ -1,9 +1,43 @@
-from fastapi import FastAPI
+import time
+
+from fastapi import FastAPI, Request
+
+from .api.routers import (
+    buses,
+    contacts,
+    courses,
+    dining,
+    energy,
+    libraries,
+    locations,
+    newsletters,
+    resources,
+    rpage,
+)
 
 app = FastAPI()
 
-# 載入 API (遷移至新 API)
-from .api import *
+
+@app.middleware("http")
+async def add_process_time_header(request: Request, call_next):
+    start_time = time.time()
+    response = await call_next(request)
+    process_time = time.time() - start_time
+    response.headers["X-Process-Time"] = str(process_time)
+    return response
+
+
+app.include_router(buses.router, prefix="/buses", tags=["Buses"])
+app.include_router(courses.router, prefix="/courses", tags=["Courses"])
+app.include_router(dining.router, prefix="/dining", tags=["Dining"])
+app.include_router(energy.router, prefix="/energy", tags=["Energy"])
+app.include_router(libraries.router, prefix="/libraries", tags=["Libraries"])
+app.include_router(locations.router, prefix="/locations", tags=["Locations"])
+app.include_router(newsletters.router, prefix="/newsletters", tags=["Newsletters"])
+app.include_router(contacts.router, prefix="/phones", tags=["Phones"])
+app.include_router(contacts.router, prefix="/contacts", tags=["Contacts"])
+app.include_router(resources.router, prefix="/resources", tags=[])
+app.include_router(rpage.router, prefix="/rpage", tags=["Rpage"])
 
 
 @app.get(
@@ -17,7 +51,7 @@
         },
     },
 )
-def home():
+async def home():
     return {"message": "Hello World!"}
 
 
@@ -32,5 +66,5 @@ def home():
         },
     },
 )
-def ping():
+async def ping():
     return {"message": "pong"}

src/api/__init__.py

@@ -1,25 +0,0 @@
-from .. import app
-from .routers import (
-    buses,
-    contacts,
-    courses,
-    dining,
-    energy,
-    libraries,
-    locations,
-    newsletters,
-    resources,
-    rpage,
-)
-
-app.include_router(buses.router, prefix="/buses", tags=["Buses"])
-app.include_router(courses.router, prefix="/courses", tags=["Courses"])
-app.include_router(dining.router, prefix="/dining", tags=["Dining"])
-app.include_router(energy.router, prefix="/energy", tags=["Energy"])
-app.include_router(libraries.router, prefix="/libraries", tags=["Libraries"])
-app.include_router(locations.router, prefix="/locations", tags=["Locations"])
-app.include_router(newsletters.router, prefix="/newsletters", tags=["Newsletters"])
-app.include_router(contacts.router, prefix="/phones", tags=["Phones"])
-app.include_router(contacts.router, prefix="/contacts", tags=["Contacts"])
-app.include_router(resources.router, prefix="/resources", tags=[])
-app.include_router(rpage.router, prefix="/rpage", tags=["Rpage"])

src/api/models/courses.py

@@ -1,8 +1,8 @@
 import json
+import operator
 import re
 
-import requests
-from cachetools import TTLCache, cached
+from src.utils import cached_request
 
 
 class CoursesData:
@@ -84,9 +84,9 @@ def check(self, course: CoursesData) -> bool:
         course_data_dict = vars(course)
         field_data = course_data_dict[self.row_field]
 
-        if self.regex_match == True:
+        if self.regex_match is True:
             match_res = re.search(self.matcher, field_data)
-            return False if match_res == None else True
+            return False if match_res is None else True
         else:
             return field_data == self.matcher
 
@@ -109,7 +109,7 @@ def __init__(
         *,
         list_build_target: list = None,
     ) -> None:
-        if list_build_target != None:
+        if list_build_target is not None:
             # 優先使用 list_build_target 建立條件式
             self.condition_stat = list_build_target
         else:
@@ -137,29 +137,29 @@ def _solve_condition_stat(self, data: dict) -> bool:
         """遞迴函式，拆分成 左手邊、運算子、右手邊，將左右手遞迴解成 ``bool`` 之後，再算出這一層的結果。"""
         # 這部分遞迴實作成這樣，是因為 pydantic 幫忙確認過條件式每一層的結構，
         # 都是合乎 [Condition, op, Condition] 的，不會有其他不合法的結構，所以可以這樣寫
-
         lhs, op, rhs = data
 
-        if type(lhs) == dict:
-            lhs = Condition(**lhs).check(self.course)
-        elif type(lhs) == list:
-            # nested Condition，可以再遞迴拆解
-            lhs = self._solve_condition_stat(lhs)
-        elif type(lhs) == Condition:
-            lhs = lhs.check(self.course)
-
-        if type(rhs) == dict:
-            rhs = Condition(**rhs).check(self.course)
-        elif type(rhs) == list:
+        lhs = self._check_condition(lhs)
+        rhs = self._check_condition(rhs)
+
+        match op:
+            case "and":
+                return lhs and rhs
+            case "or":
+                return lhs or rhs
+            case _:
+                raise ValueError(f"Unknown operator: {op}")
+
+    def _check_condition(self, item):
+        if isinstance(item, dict):
+            return Condition(**item).check(self.course)
+        elif isinstance(item, list):
             # nested Condition，可以再遞迴拆解
-            rhs = self._solve_condition_stat(rhs)
-        elif type(rhs) == Condition:
-            rhs = rhs.check(self.course)
-
-        if op == "and":
-            return lhs and rhs
-        elif op == "or":
-            return lhs or rhs
+            return self._solve_condition_stat(item)
+        elif isinstance(item, Condition):
+            return item.check(self.course)
+        else:
+            return item
 
     def accept(self, course: CoursesData) -> bool:
         """包裝遞迴函式供外部使用，回傳以該課程計算多個條件運算後的結果。"""
@@ -182,17 +182,16 @@ class Processor:
     def __init__(self, json_path=None) -> None:
         self.course_data = self._get_course_data(json_path)
 
-    @cached(cache=TTLCache(maxsize=1, ttl=60 * 60))
     def _get_course_data(self, json_path=None) -> list[CoursesData]:
         """TODO: error handler."""
-        if json_path != None:
+        if json_path is not None:
             # 使用 json 模組讀取檔案
             with open(json_path, "r", encoding="utf-8") as f:
                 course_data_dict_list = json.load(f)
         else:
             # 使用 requests 模組取得網頁資料
-            course_data_resp = requests.get(self.NTHU_COURSE_DATA_URL)
-            course_data_dict_list = json.loads(course_data_resp.text)
+            course_data_resp = cached_request.get(self.NTHU_COURSE_DATA_URL)
+            course_data_dict_list = json.loads(course_data_resp)
         return list(map(CoursesData, course_data_dict_list))
 
     def update(self, json_path=None):
@@ -213,28 +212,21 @@ def list_selected_fields(self, field) -> list:
         fields_list = list(set(fields_list))
         return fields_list
 
-    def list_credit(self, credit: float, op: str = None) -> list:
-        res = []
-        if op == None or op == "":
-            res = [
-                course for course in self.course_data if float(course.credit) == credit
-            ]
-        elif op == "gt":
-            res = [
-                course for course in self.course_data if float(course.credit) > credit
-            ]
-        elif op == "lt":
-            res = [
-                course for course in self.course_data if float(course.credit) < credit
-            ]
-        elif op == "gte":
-            res = [
-                course for course in self.course_data if float(course.credit) >= credit
-            ]
-        elif op == "lte":
-            res = [
-                course for course in self.course_data if float(course.credit) <= credit
-            ]
+    def list_credit(self, credit: float, op: str = "") -> list:
+        ops = {
+            "gt": operator.gt,
+            "lt": operator.lt,
+            "gte": operator.ge,
+            "lte": operator.le,
+            "eq": operator.eq,
+            "": operator.eq,
+        }
+
+        res = [
+            course
+            for course in self.course_data
+            if ops[op](float(course.credit), credit)
+        ]
 
         return res
 

src/api/models/dining.py

@@ -10,7 +10,7 @@ class Dining:
     # 餐廳及服務性廠商
     # https://ddfm.site.nthu.edu.tw/p/404-1494-256455.php?Lang=zh-tw
 
-    def get_dining_data(self) -> dict:
+    def get_dining_data(self) -> list:
         url = "https://ddfm.site.nthu.edu.tw/p/404-1494-256455.php?Lang=zh-tw"
         res_text = cached_request.get(url)
         # 將字串轉換成 json 格式
@@ -20,14 +20,14 @@ def get_dining_data(self) -> dict:
         if dining_data is not None:
             dining_data = dining_data.group(1)
         else:
-            return {}
+            return []
         dining_data = dining_data.replace("'", '"')
         dining_data = dining_data.replace("\n", "")
         dining_data = dining_data.replace(",  ]", "]")
         dining_data = json.loads(dining_data)
         return dining_data
 
-    def get_all_building_names(self) -> list[str]:
+    def get_all_building_names(self) -> list:
         dining_data = self.get_dining_data()
         building_names = []
         for building in dining_data:
@@ -47,7 +47,7 @@ def query_by_building_name(self, query_name):
         for restaurant in dining_data:
             if restaurant["building"] == query_name:
                 return restaurant
-        return None
+        return {}
 
     def query_by_restaurant_name(self, query_name):
         dining_data = self.get_dining_data()

src/api/routers/contacts.py

@@ -1,4 +1,3 @@
-from doctest import Example
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, Path