Coverage for src/data_api/domain/courses/models.py: 98%

1""" 

2Courses domain models. 

3 

4Typed course predicates, independent of HTTP transports. 

5""" 

6 

7import re 

8from collections.abc import Callable 

9from dataclasses import dataclass, field 

10from functools import cached_property 

11from typing import Any, Optional, Union 

12 

13from pydantic import BaseModel 

14 

15from data_api.models.courses import CourseData, CourseMatcher, validate_matcher 

16 

17 

18def field_values(value: Any, path: list[str]) -> list[CourseMatcher]: 

19 """Resolve array leaves without confusing absent union keys with explicit null.""" 

20 if isinstance(value, list): 

21 return [leaf for item in value for leaf in field_values(item, path)] 

22 if not path: 

23 return [value] 

24 key, *rest = path 

25 if isinstance(value, BaseModel): 

26 if key not in type(value).model_fields: 

27 return [] 

28 return field_values(getattr(value, key), rest) 

29 return [] 

30 

31 

32@dataclass 

33class Condition: 

34 """Single condition for filtering courses.""" 

35 

36 row_field: str 

37 matcher: CourseMatcher 

38 regex_match: bool = False 

39 _pattern: re.Pattern | None = field(default=None, init=False, repr=False) 

40 

41 def __post_init__(self): 

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

43 if self.regex_match and isinstance(self.matcher, str): 

44 self._pattern = re.compile(self.matcher) 

45 

46 def check(self, course: CourseData) -> bool: 

47 """Check if course satisfies this condition.""" 

48 for value in field_values(course, self.row_field.split(".")): 

49 if self._pattern is not None: 

50 if isinstance(value, str) and self._pattern.search(value) is not None: 

51 return True 

52 elif value == self.matcher: 

53 return True 

54 return False 

55 

56 

57ConditionType = Union[Condition, bool, dict, list] 

58 

59 

60def validate_condition_sequence( 

61 sequence: list[Any], 

62 *, 

63 operand_types: tuple[type, ...] = (Condition, bool, dict, list), 

64) -> None: 

65 """Validate alternating operands and operators without evaluating conditions.""" 

66 if sequence and len(sequence) % 2 == 0: 

67 raise ValueError("Condition arrays must have an odd number of elements.") 

68 for index, item in enumerate(sequence): 

69 if index % 2 == 0: 

70 if not isinstance(item, operand_types): 

71 raise ValueError("Expected a condition object or nested array at even positions.") 

72 elif not isinstance(item, str) or item not in ("and", "or"): 

73 raise ValueError("Unknown operator; expected 'and' or 'or' at odd positions.") 

74 

75 

76@dataclass 

77class Conditions: 

78 """Complex condition tree for filtering courses.""" 

79 

80 condition_stat: Any = field(default=None) 

81 

82 def __init__( 

83 self, 

84 row_field: Optional[str] = None, 

85 matcher: CourseMatcher = None, 

86 regex_match: bool = False, 

87 *, 

88 list_build_target: Optional[list[Any]] = None, 

89 ): 

90 if list_build_target is not None: 

91 self.condition_stat = list_build_target 

92 elif row_field is not None: 

93 self.condition_stat = [ 

94 Condition(row_field, matcher, regex_match), 

95 "and", 

96 True, 

97 ] 

98 else: 

99 raise ValueError("Must provide row_field and matcher, or list_build_target") 

100 

101 def __and__(self, other: "Conditions") -> "Conditions": 

102 return Conditions(list_build_target=[self.condition_stat, "and", other.condition_stat]) 

103 

104 def __or__(self, other: "Conditions") -> "Conditions": 

105 return Conditions(list_build_target=[self.condition_stat, "or", other.condition_stat]) 

106 

107 @staticmethod 

108 def _prepare_sequence(item: list[Any]) -> Callable[[CourseData], bool]: 

109 validate_condition_sequence(item) 

110 if not item: 

111 return lambda course: True 

112 operands = [Conditions._prepare(value) for value in item[::2]] 

113 operators = tuple(item[1::2]) 

114 

115 def evaluate(course: CourseData) -> bool: 

116 result = operands[0](course) 

117 for operator, predicate in zip(operators, operands[1:], strict=True): 

118 value = predicate(course) 

119 result = result and value if operator == "and" else result or value 

120 return result 

121 

122 return evaluate 

123 

124 @staticmethod 

125 def _prepare(item: ConditionType) -> Callable[[CourseData], bool]: 

126 if isinstance(item, dict): 

127 return Condition(**item).check 

128 if isinstance(item, list): 

129 return Conditions._prepare_sequence(item) 

130 if isinstance(item, Condition): 

131 return item.check 

132 if isinstance(item, bool): 

133 return lambda course: item 

134 raise TypeError(f"Cannot handle condition item: {item}") 

135 

136 @cached_property 

137 def _predicate(self) -> Callable[[CourseData], bool]: 

138 return self._prepare(self.condition_stat) 

139 

140 def accept(self, course: CourseData) -> bool: 

141 """Check if course satisfies all conditions.""" 

142 return self._predicate(course)