Coverage for src/data_api/domain/courses/services.py: 100%

1"""Normalized, independently cached course snapshots for each published semester.""" 

2 

3from pydantic import TypeAdapter 

4 

5from data_api.core.exceptions import DataNotAvailableException 

6from data_api.data.manager import nthudata 

7from data_api.data.nthudata import ( 

8 DatasetState, 

9 FetchFailure, 

10 Freshness, 

11 JsonData, 

12 NTHUDataManager, 

13 Snapshot, 

14) 

15from data_api.domain.courses.models import Conditions 

16from data_api.models.courses import CourseCatalog, CourseData, CourseListName, SemesterCode 

17 

18LATEST_PATH = "courses/latest.json" 

19CATALOG_PATH = "courses/sources.json" 

20 

21 

22class UnknownSemester(Exception): 

23 """A well-formed semester absent from a verified published catalog.""" 

24 

25 

26def prepare_courses(raw: JsonData, semester: str | None = None) -> list[CourseData]: 

27 if not isinstance(raw, list): 

28 raise FetchFailure("payload_type") 

29 courses = TypeAdapter(list[CourseData]).validate_python(raw, strict=True) 

30 ids = [course.id for course in courses] 

31 if len(ids) != len(set(ids)) or ( 

32 semester is not None and any(course.semester_code != semester for course in courses) 

33 ): 

34 raise FetchFailure("validation") 

35 return courses 

36 

37 

38def prepare_catalog(raw: JsonData) -> list[str]: 

39 return CourseCatalog.model_validate(raw, strict=True).semesters 

40 

41 

42class CoursesService: 

43 def __init__(self, manager: NTHUDataManager = nthudata) -> None: 

44 self.manager = manager 

45 self.state = manager.register(LATEST_PATH, prepare_courses) 

46 self.catalog_state = manager.register(CATALOG_PATH, prepare_catalog) 

47 self._semesters: dict[str, DatasetState[list[CourseData]]] = {} 

48 

49 async def get_semesters(self) -> Snapshot[list[str]]: 

50 return await self.manager.get_snapshot(CATALOG_PATH, self.catalog_state) 

51 

52 async def get_snapshot(self, semester: str | None = None) -> Snapshot[list[CourseData]]: 

53 if semester is None: 

54 return await self.manager.get_snapshot(LATEST_PATH, self.state) 

55 semester = TypeAdapter(SemesterCode).validate_python(semester, strict=True) 

56 catalog = await self.get_semesters() 

57 if semester not in catalog.data: 

58 if self.catalog_state.freshness != Freshness.CURRENT: 

59 raise DataNotAvailableException("Semester catalog temporarily unavailable.") 

60 raise UnknownSemester("Semester is not published.") 

61 path = f"courses/semesters/{semester}.json" 

62 if semester not in self._semesters: 

63 self._semesters[semester] = self.manager.register( 

64 path, lambda raw: prepare_courses(raw, semester) 

65 ) 

66 return await self.manager.get_snapshot(path, self._semesters[semester]) 

67 

68 @staticmethod 

69 def query( 

70 courses: list[CourseData], 

71 conditions: Conditions, 

72 course_type: CourseListName | None = None, 

73 ) -> list[CourseData]: 

74 return [ 

75 course 

76 for course in courses 

77 if conditions.accept(course) 

78 and ( 

79 course_type is None 

80 or ( 

81 course_type == CourseListName.microcredits 

82 and course.credits is not None 

83 and not course.credits.is_integer() 

84 ) 

85 or (course_type == CourseListName.xclass and "X-Class" in course.note) 

86 ) 

87 ] 

88 

89 

90courses_service = CoursesService()