Coverage for src/data_api/models/courses.py: 100%

201 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-11-13 19:28 +0000

1"""Shared normalized course data and typed query models.""" 

2 

3import math 

4import re 

5from enum import StrEnum 

6from typing import Annotated, Any, Self 

7 

8from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator, model_validator 

9 

10COURSE_QUERY_MAX_DEPTH = 32 

11COURSE_QUERY_MAX_NODES = 1024 

12SemesterCode = Annotated[str, Field(pattern=r"^[0-9]{5}$")] 

13NonnegativeInt = Annotated[int, Field(strict=True, ge=0)] 

14Credits = Annotated[float, Field(strict=True, ge=0, allow_inf_nan=False)] 

15CourseMatcher = str | int | float | bool | None 

16 

17 

18class CourseLanguage(StrEnum): 

19 Chinese = "zh" 

20 English = "en" 

21 

22 

23class LecturerLabel(BaseModel): 

24 model_config = ConfigDict(strict=True, extra="forbid") 

25 label: str | None 

26 

27 

28class LecturerNames(BaseModel): 

29 model_config = ConfigDict(strict=True, extra="forbid") 

30 name_zh: str | None 

31 name_en: str | None 

32 

33 

34class CourseMeeting(BaseModel): 

35 model_config = ConfigDict(strict=True) 

36 classroom: str | None 

37 time: str | None 

38 time_codes: list[str] | None 

39 

40 

41class CourseExpertise(BaseModel): 

42 model_config = ConfigDict(strict=True) 

43 name: str | None 

44 type: str | None 

45 

46 

47class CurriculumRequirement(BaseModel): 

48 model_config = ConfigDict(strict=True) 

49 target: str | None 

50 requirement: str | None 

51 

52 

53class CourseData(BaseModel): 

54 """A published normalized course, including unknown values and parser warnings.""" 

55 

56 model_config = ConfigDict(strict=True) 

57 

58 id: str = Field(min_length=1, description="課號") 

59 chinese_title: str 

60 english_title: str 

61 note: str 

62 limit_note: str 

63 object: str = Field(description="開課對象/選課限制") 

64 ge_type: str = Field(description="通識對象") 

65 ge_category: str = Field(description="通識分類") 

66 no_extra_selection: str 

67 semester_code: SemesterCode 

68 department_code: str | None 

69 course_number: str | None 

70 department_name: str | None 

71 college_name: str | None 

72 course_level_code: str | None 

73 course_level: str | None 

74 remote_teaching: str | None 

75 credits: Credits | None 

76 capacity: NonnegativeInt | None 

77 capacity_unlimited: bool | None 

78 enrolled_count: NonnegativeInt | None 

79 enrollment_after_selection: NonnegativeInt | None 

80 enrollment_after_withdrawal: NonnegativeInt | None 

81 reserved_for_freshmen: NonnegativeInt | None 

82 is_cancelled: bool | None 

83 language_code: str | None = Field(pattern=r"^(zh|en)$") 

84 lecturers: list[LecturerLabel | LecturerNames] 

85 meetings: list[CourseMeeting] 

86 expertises: list[CourseExpertise] 

87 programs: list[str] 

88 curriculum_requirements: list[CurriculumRequirement] 

89 prerequisite_text: str 

90 parsing_issues: list[str] 

91 

92 @model_validator(mode="after") 

93 def check_identity(self) -> Self: 

94 if not self.id.startswith(self.semester_code) or not self.id[5:].strip(): 

95 raise ValueError("Course ID must start with its semester and contain a course identity") 

96 if not (self.chinese_title.strip() or self.english_title.strip()): 

97 raise ValueError("A course must have a usable title") 

98 return self 

99 

100 

101class CourseCatalog(BaseModel): 

102 model_config = ConfigDict(strict=True) 

103 semesters: list[SemesterCode] 

104 

105 @field_validator("semesters") 

106 @classmethod 

107 def unique_semesters(cls, value: list[str]) -> list[str]: 

108 if len(value) != len(set(value)): 

109 raise ValueError("Duplicate semesters") 

110 return sorted(value, reverse=True) 

111 

112 

113class CourseFieldName(StrEnum): 

114 id = "id" 

115 chinese_title = "chinese_title" 

116 english_title = "english_title" 

117 note = "note" 

118 limit_note = "limit_note" 

119 object = "object" 

120 ge_type = "ge_type" 

121 ge_category = "ge_category" 

122 no_extra_selection = "no_extra_selection" 

123 semester_code = "semester_code" 

124 department_code = "department_code" 

125 course_number = "course_number" 

126 department_name = "department_name" 

127 college_name = "college_name" 

128 course_level_code = "course_level_code" 

129 course_level = "course_level" 

130 remote_teaching = "remote_teaching" 

131 credits = "credits" 

132 capacity = "capacity" 

133 capacity_unlimited = "capacity_unlimited" 

134 enrolled_count = "enrolled_count" 

135 enrollment_after_selection = "enrollment_after_selection" 

136 enrollment_after_withdrawal = "enrollment_after_withdrawal" 

137 reserved_for_freshmen = "reserved_for_freshmen" 

138 is_cancelled = "is_cancelled" 

139 language_code = "language_code" 

140 lecturer_label = "lecturers.label" 

141 lecturer_name_zh = "lecturers.name_zh" 

142 lecturer_name_en = "lecturers.name_en" 

143 meeting_classroom = "meetings.classroom" 

144 meeting_time = "meetings.time" 

145 meeting_time_codes = "meetings.time_codes" 

146 expertise_name = "expertises.name" 

147 expertise_type = "expertises.type" 

148 programs = "programs" 

149 requirement_target = "curriculum_requirements.target" 

150 requirement = "curriculum_requirements.requirement" 

151 prerequisite_text = "prerequisite_text" 

152 parsing_issues = "parsing_issues" 

153 

154 

155INTEGER_FIELDS = { 

156 CourseFieldName.capacity.value, 

157 CourseFieldName.enrolled_count.value, 

158 CourseFieldName.enrollment_after_selection.value, 

159 CourseFieldName.enrollment_after_withdrawal.value, 

160 CourseFieldName.reserved_for_freshmen.value, 

161} 

162BOOLEAN_FIELDS = { 

163 CourseFieldName.capacity_unlimited.value, 

164 CourseFieldName.is_cancelled.value, 

165} 

166NUMERIC_OR_BOOLEAN_FIELDS = INTEGER_FIELDS | BOOLEAN_FIELDS | {CourseFieldName.credits.value} 

167NULLABLE_FIELDS = NUMERIC_OR_BOOLEAN_FIELDS | { 

168 CourseFieldName.department_code.value, 

169 CourseFieldName.course_number.value, 

170 CourseFieldName.department_name.value, 

171 CourseFieldName.college_name.value, 

172 CourseFieldName.course_level_code.value, 

173 CourseFieldName.course_level.value, 

174 CourseFieldName.remote_teaching.value, 

175 CourseFieldName.language_code.value, 

176 CourseFieldName.lecturer_label.value, 

177 CourseFieldName.lecturer_name_zh.value, 

178 CourseFieldName.lecturer_name_en.value, 

179 CourseFieldName.meeting_classroom.value, 

180 CourseFieldName.meeting_time.value, 

181 CourseFieldName.meeting_time_codes.value, 

182 CourseFieldName.expertise_name.value, 

183 CourseFieldName.expertise_type.value, 

184 CourseFieldName.requirement_target.value, 

185 CourseFieldName.requirement.value, 

186} 

187 

188 

189def _is_valid_credits_matcher(matcher: CourseMatcher) -> bool: 

190 if type(matcher) is int: 

191 return matcher >= 0 

192 if type(matcher) is float: 

193 return math.isfinite(matcher) and matcher >= 0 

194 return False 

195 

196 

197def _is_valid_matcher_value(field: str, matcher: CourseMatcher) -> bool: 

198 if matcher is None: 

199 return field in NULLABLE_FIELDS 

200 if field in BOOLEAN_FIELDS: 

201 return type(matcher) is bool 

202 if field in INTEGER_FIELDS: 

203 return type(matcher) is int and matcher >= 0 

204 if field == CourseFieldName.credits.value: 

205 return _is_valid_credits_matcher(matcher) 

206 return isinstance(matcher, str) 

207 

208 

209def validate_matcher(field: str, matcher: CourseMatcher, regex_match: bool) -> None: 

210 """Validate types before querying, including empty datasets.""" 

211 CourseFieldName(field) 

212 if not _is_valid_matcher_value(field, matcher): 

213 raise ValueError(f"Invalid matcher type for course field '{field}'") 

214 if not regex_match: 

215 return 

216 if not isinstance(matcher, str) or field in NUMERIC_OR_BOOLEAN_FIELDS: 

217 raise ValueError("Regular expressions require a text field and string matcher") 

218 try: 

219 re.compile(matcher) 

220 except (re.error, OverflowError, RecursionError) as exc: 

221 raise ValueError("Invalid regular expression") from exc 

222 

223 

224class CourseSearchParams(BaseModel): 

225 """AND filters: regex text, exact numbers/booleans, ANY over array leaves.""" 

226 

227 model_config = ConfigDict(extra="forbid") 

228 

229 id: str | None = None 

230 chinese_title: str | None = None 

231 english_title: str | None = None 

232 note: str | None = None 

233 limit_note: str | None = None 

234 object: str | None = None 

235 ge_type: str | None = None 

236 ge_category: str | None = None 

237 no_extra_selection: str | None = None 

238 semester_code: str | None = None 

239 department_code: str | None = None 

240 course_number: str | None = None 

241 department_name: str | None = None 

242 college_name: str | None = None 

243 course_level_code: str | None = None 

244 course_level: str | None = None 

245 remote_teaching: str | None = None 

246 credits: Annotated[float, Field(ge=0, allow_inf_nan=False)] | None = None 

247 capacity: Annotated[int, Field(ge=0)] | None = None 

248 capacity_unlimited: bool | None = None 

249 enrolled_count: Annotated[int, Field(ge=0)] | None = None 

250 enrollment_after_selection: Annotated[int, Field(ge=0)] | None = None 

251 enrollment_after_withdrawal: Annotated[int, Field(ge=0)] | None = None 

252 reserved_for_freshmen: Annotated[int, Field(ge=0)] | None = None 

253 is_cancelled: bool | None = None 

254 language_code: str | None = None 

255 lecturer_label: str | None = Field(None, alias=CourseFieldName.lecturer_label.value) 

256 lecturer_name_zh: str | None = Field(None, alias=CourseFieldName.lecturer_name_zh.value) 

257 lecturer_name_en: str | None = Field(None, alias=CourseFieldName.lecturer_name_en.value) 

258 meeting_classroom: str | None = Field(None, alias=CourseFieldName.meeting_classroom.value) 

259 meeting_time: str | None = Field(None, alias=CourseFieldName.meeting_time.value) 

260 meeting_time_codes: str | None = Field(None, alias=CourseFieldName.meeting_time_codes.value) 

261 expertise_name: str | None = Field(None, alias=CourseFieldName.expertise_name.value) 

262 expertise_type: str | None = Field(None, alias=CourseFieldName.expertise_type.value) 

263 programs: str | None = None 

264 requirement_target: str | None = Field(None, alias=CourseFieldName.requirement_target.value) 

265 requirement: str | None = Field(None, alias=CourseFieldName.requirement.value) 

266 prerequisite_text: str | None = None 

267 parsing_issues: str | None = None 

268 

269 

270class CourseCondition(BaseModel): 

271 model_config = ConfigDict(strict=True) 

272 row_field: CourseFieldName 

273 matcher: CourseMatcher 

274 regex_match: bool = False 

275 

276 @field_validator("row_field", mode="before") 

277 @classmethod 

278 def parse_field(cls, value: Any) -> CourseFieldName: 

279 return CourseFieldName(value) 

280 

281 @model_validator(mode="after") 

282 def check_matcher(self) -> Self: 

283 validate_matcher(self.row_field, self.matcher, self.regex_match) 

284 return self 

285 

286 

287class CourseQueryOperation(StrEnum): 

288 and_ = "and" 

289 or_ = "or" 

290 

291 

292class CourseQueryCondition(RootModel): 

293 """Alternate conditions and and/or, evaluated left to right; [] matches all. 

294 

295 Maximum 32 array levels and 1024 nodes (arrays, condition objects and operators). 

296 """ 

297 

298 root: list["CourseQueryCondition | CourseCondition | CourseQueryOperation"] 

299 

300 @field_validator("root", mode="before") 

301 @classmethod 

302 def check_query(cls, value: Any) -> Any: 

303 from data_api.domain.courses.models import validate_condition_sequence 

304 

305 if not isinstance(value, list): 

306 return value 

307 pending = [(value, 1)] 

308 nodes = 0 

309 while pending: 

310 item, depth = pending.pop() 

311 nodes += 1 

312 if isinstance(item, CourseQueryCondition): 

313 item = item.root 

314 if not isinstance(item, list): 

315 continue 

316 if depth > COURSE_QUERY_MAX_DEPTH: 

317 raise ValueError( 

318 f"Course queries cannot exceed {COURSE_QUERY_MAX_DEPTH} array levels." 

319 ) 

320 if nodes + len(pending) + len(item) > COURSE_QUERY_MAX_NODES: 

321 raise ValueError(f"Course queries cannot exceed {COURSE_QUERY_MAX_NODES} nodes.") 

322 validate_condition_sequence( 

323 item, operand_types=(list, dict, CourseCondition, CourseQueryCondition) 

324 ) 

325 pending.extend((child, depth + 1) for child in item) 

326 return value 

327 

328 

329class CourseListName(StrEnum): 

330 microcredits = "microcredits" 

331 xclass = "xclass" 

332 

333 

334class CourseGetParams(CourseSearchParams): 

335 type: CourseListName | None = None