Coverage for src/data_api/api/routers/courses.py: 100%

1"""Latest and semester-specific normalized course queries.""" 

2 

3from typing import Annotated 

4 

5from fastapi import APIRouter, Body, Depends, HTTPException, Query, Response 

6from pydantic import ValidationError 

7 

8from data_api.api.schemas import courses as schemas 

9from data_api.api.schemas.errors import VALIDATION_ERROR_RESPONSES, ErrorResponse 

10from data_api.api.schemas.responses import COUNTED_SNAPSHOT_RESPONSES 

11from data_api.data.nthudata import Snapshot 

12from data_api.domain.courses.models import Conditions 

13from data_api.domain.courses.services import UnknownSemester 

14from data_api.models.courses import SemesterCode 

15from data_api.runtime import get_runtime 

16 

17router = APIRouter(responses={**COUNTED_SNAPSHOT_RESPONSES, **VALIDATION_ERROR_RESPONSES}) 

18COURSE_QUERY_ERROR_RESPONSES = { 

19 422: { 

20 **VALIDATION_ERROR_RESPONSES[422], 

21 "description": "Invalid condition grammar, matcher types, query limits or regular expression", 

22 } 

23} 

24SEMESTER_ERRORS = {404: {"model": ErrorResponse, "description": "Semester is not published"}} 

25 

26 

27async def latest_snapshot() -> Snapshot[list[schemas.CourseData]]: 

28 return await get_runtime().courses.get_snapshot() 

29 

30 

31async def semester_snapshot(semester: SemesterCode) -> Snapshot[list[schemas.CourseData]]: 

32 try: 

33 return await get_runtime().courses.get_snapshot(semester) 

34 except UnknownSemester as exc: 

35 raise HTTPException(404, detail="Semester is not published.") from exc 

36 

37 

38LatestSnapshot = Annotated[Snapshot[list[schemas.CourseData]], Depends(latest_snapshot)] 

39SemesterSnapshot = Annotated[Snapshot[list[schemas.CourseData]], Depends(semester_snapshot)] 

40CourseFilters = Annotated[schemas.CourseGetParams, Query()] 

41CourseQuery = Annotated[schemas.CourseQueryCondition | schemas.CourseCondition, Body()] 

42 

43 

44def counted_response[T]( 

45 response: Response, snapshot: Snapshot[list[T]], result: list[T] 

46) -> list[T]: 

47 response.headers["X-Total-Count"] = str(len(result)) 

48 if snapshot.version is not None: 

49 response.headers["X-Data-Commit-Hash"] = snapshot.version 

50 return result 

51 

52 

53def filter_courses( 

54 response: Response, 

55 snapshot: Snapshot[list[schemas.CourseData]], 

56 filters: schemas.CourseGetParams, 

57) -> list[schemas.CourseData]: 

58 condition = Conditions(list_build_target=[]) 

59 for field_name, value in filters.model_dump( 

60 mode="json", by_alias=True, exclude_none=True, exclude={"type"} 

61 ).items(): 

62 if value == "": 

63 continue 

64 try: 

65 query = schemas.CourseCondition( 

66 row_field=field_name, matcher=value, regex_match=isinstance(value, str) 

67 ) 

68 except ValidationError as exc: 

69 raise HTTPException( 

70 422, detail=f"Invalid regular expression for course field '{field_name}'." 

71 ) from exc 

72 condition &= Conditions(query.row_field, query.matcher, query.regex_match) 

73 result = get_runtime().courses.query(snapshot.data, condition, filters.type) 

74 return counted_response(response, snapshot, result) 

75 

76 

77def query_courses( 

78 response: Response, 

79 snapshot: Snapshot[list[schemas.CourseData]], 

80 query: schemas.CourseQueryCondition | schemas.CourseCondition, 

81) -> list[schemas.CourseData]: 

82 body = query.model_dump(mode="json") 

83 condition = Conditions(list_build_target=[body] if isinstance(body, dict) else body) 

84 result = get_runtime().courses.query(snapshot.data, condition) 

85 return counted_response(response, snapshot, result) 

86 

87 

88@router.get( 

89 "", 

90 response_model=list[schemas.CourseData], 

91 operation_id="getCourses", 

92 responses=VALIDATION_ERROR_RESPONSES, 

93) 

94async def get_courses(response: Response, filters: CourseFilters, snapshot: LatestSnapshot): 

95 """查詢最新課程。文字欄位使用 regex，數值與布林精確比對；所有條件以 AND 組合。""" 

96 return filter_courses(response, snapshot, filters) 

97 

98 

99@router.get("/semesters", response_model=list[SemesterCode], operation_id="getCourseSemesters") 

100async def get_semesters(response: Response): 

101 """列出已公布的學期碼，由新到舊排序。""" 

102 snapshot = await get_runtime().courses.get_semesters() 

103 return counted_response(response, snapshot, snapshot.data) 

104 

105 

106@router.post( 

107 "/query", 

108 response_model=list[schemas.CourseData], 

109 operation_id="queryCourses", 

110 responses=COURSE_QUERY_ERROR_RESPONSES, 

111) 

112async def search_courses_by_condition( 

113 response: Response, query_condition: CourseQuery, snapshot: LatestSnapshot 

114): 

115 """進階搜尋最新課程。條件以 and/or 交替並由左至右計算，支援巢狀陣列。 

116 

117 空陣列符合全部；最多 32 層與 1024 節點。matcher 使用欄位的原生型別， 

118 regex_match 只適用文字，預設為 false。陣列葉節點採任一元素符合。 

119 """ 

120 return query_courses(response, snapshot, query_condition) 

121 

122 

123@router.get( 

124 "/{semester}", 

125 response_model=list[schemas.CourseData], 

126 operation_id="getSemesterCourses", 

127 responses={**SEMESTER_ERRORS, **VALIDATION_ERROR_RESPONSES}, 

128) 

129async def get_semester_courses( 

130 response: Response, filters: CourseFilters, snapshot: SemesterSnapshot 

131): 

132 """查詢指定學期課程，篩選方式與 GET /courses 相同，不自動改查其他學期。""" 

133 return filter_courses(response, snapshot, filters) 

134 

135 

136@router.post( 

137 "/{semester}/query", 

138 response_model=list[schemas.CourseData], 

139 operation_id="querySemesterCourses", 

140 responses={**SEMESTER_ERRORS, **COURSE_QUERY_ERROR_RESPONSES}, 

141) 

142async def query_semester_courses( 

143 response: Response, query_condition: CourseQuery, snapshot: SemesterSnapshot 

144): 

145 """進階搜尋指定學期，條件文法與 POST /courses/query 相同。""" 

146 return query_courses(response, snapshot, query_condition)