fix: ignore /openai.json and add match_field func

.flake8

@@ -1,4 +1,4 @@
 [flake8]
-ignore = E501
+ignore = E501,W503
 exclude = .git,__pycache__,web,.venv
 max-complexity = 10

src/backend/dao.py

@@ -1,21 +1,48 @@
+import logging
 from typing import List
 
-from sqlalchemy import func
+from sqlalchemy import Column, func
 from sqlalchemy.orm import Session
 
 from .models import Job
 
 
+def match_field(field: str) -> Column[str]:
+    match field:
+        case "city":
+            return Job.city
+        case "position":
+            return Job.position
+        case "education":
+            return Job.education
+        case "experience":
+            return Job.experience
+        case "company_name":
+            return Job.company_name
+        case "industry":
+            return Job.industry
+        case "financing_stage":
+            return Job.financing_stage
+        case "company_size":
+            return Job.company_size
+        case "welfare":
+            return Job.welfare
+        case _:
+            logging.error("Invalid field name")
+            return Job.city
+
+
 def get(session: Session, offset=0, limit=500) -> List[Job]:
     result = session.query(Job).offset(offset).limit(limit).all()
     return result
 
 
-def group_count(session: Session, key: str) -> dict:
+def group_count(session: Session, key: str) -> dict[str, int]:
     valid_keys = ["city", "education"]
     if key not in valid_keys:
         return {}
-    group_by_column = getattr(Job, key)
+    group_by_column = match_field(key)
+
     result = (
         session.query(group_by_column, func.count(Job.id))
         .group_by(group_by_column)
@@ -27,9 +54,15 @@ def group_count(session: Session, key: str) -> dict:
 
 # select : exist in the target column
 def existed_select(
-    session: Session, field: str = "city", exist: str = "成都",
-    offset=0, limit=500
+    session: Session, field: str = "city", exist: str = "成都", offset=0, limit=100
 ) -> list[Job]:
-    tar_attr = getattr(Job, field)
-    result = session.query(Job).filter(tar_attr.like(f"%{exist}%")).offset(offset).limit(500).all()
+    tar_attr = match_field(field)
+
+    result = (
+        session.query(Job)
+        .filter(tar_attr.like(f"%{exist}%"))
+        .offset(offset)
+        .limit(limit)
+        .all()
+    )
     return result

src/backend/main.py

@@ -26,8 +26,13 @@
 
 @app.middleware("http")
 async def add_static_filter(request: Request, call_next):
-    if request.url.path.startswith("/api") or request.url.path.startswith("/docs"):
+    if (
+        request.url.path.startswith("/api")
+        or request.url.path.startswith("/docs")
+        or request.url.path == "/openapi.json"
+    ):
         return await call_next(request)
+
     if request.url.path == "/":
         return FileResponse(staticdir.joinpath("index.html"))
 

src/backend/routers.py

@@ -1,5 +1,4 @@
 from fastapi import APIRouter, Depends
-from fastapi.responses import JSONResponse
 from sqlalchemy.orm import Session
 
 from . import db, schemas, services
@@ -17,14 +16,18 @@ def get_session():
 router = APIRouter()
 
 
-@router.get("/jobs", response_model=list[schemas.JobSchema])
-def get_jobs(city: str | None = "全国", session: Session = Depends(get_session)):
+@router.get("/jobs/", response_model=list[schemas.JobSchema])
+def get_jobs(city: str | None, session: Session = Depends(get_session)):
     return services.get_jobs(city, session)
 
 
-@router.get("/analyze/{item}", response_class=JSONResponse)
+@router.get("/analyze/{item}", response_model=dict[str, int])
 def get_analyze(item: str, session: Session = Depends(get_session)):
-    return services.group_and_count(session, item)
+    match item:
+        case "city" | "education":
+            return services.group_and_count(session, item)
+        case _:
+            return {"error": 404}
 
 
 @router.get("/position/{spec}", response_model=list[schemas.SimpleJobSchema])

src/backend/services.py

@@ -4,14 +4,12 @@
 
 
 def get_jobs(city_limit: str | None, session: Session):
-    if city_limit == "全国":
+    if city_limit == "全国" or city_limit is None:
         return dao.get(session)
     return dao.existed_select(session, "city", city_limit)
 
 
 def group_and_count(session: Session, key: str):
-    if key not in ["city", "education"]:
-        return {}
     return dao.group_count(session, key)