Coverage for src/case_router/rules.py: 100%

1"""Routing rules. 

2 

3Each rule is a small object that answers one question: does this case belong in my queue? 

4A tenant's rules are stored as rows in SQL and turned into objects by ``build_rule``, so 

5every org can configure its own routing without a code change. ``RuleSet`` evaluates them 

6in order and the first match wins. 

7""" 

8 

9from __future__ import annotations 

10 

11import re 

12from abc import ABC, abstractmethod 

13from collections.abc import Callable, Iterable, Iterator, Sequence 

14 

15from .models import CaseDraft, Channel, Priority, RoutingDecision, Tier 

16 

17 

18class RoutingRule(ABC): 

19 """Base class for every rule: a target queue plus a match test.""" 

20 

21 def __init__(self, queue: str) -> None: 

22 if not queue.strip(): 

23 raise ValueError("queue must not be blank") 

24 self.queue = queue.strip() 

25 

26 @property 

27 @abstractmethod 

28 def name(self) -> str: 

29 """Short label recorded on each case and in metrics, e.g. ``tier=PREMIUM``.""" 

30 

31 @abstractmethod 

32 def matches(self, draft: CaseDraft) -> bool: 

33 """Return True when the case belongs in this rule's queue.""" 

34 

35 def __repr__(self) -> str: 

36 return f"{type(self).__name__}({self.name!r} -> {self.queue!r})" 

37 

38 

39class MinimumPriorityRule(RoutingRule): 

40 """Matches cases at or above a priority, e.g. HIGH matches HIGH and CRITICAL.""" 

41 

42 def __init__(self, minimum: Priority, queue: str) -> None: 

43 super().__init__(queue) 

44 self.minimum = minimum 

45 

46 @property 

47 def name(self) -> str: 

48 return f"priority>={self.minimum}" 

49 

50 def matches(self, draft: CaseDraft) -> bool: 

51 return draft.priority.rank >= self.minimum.rank 

52 

53 

54class TierRule(RoutingRule): 

55 def __init__(self, tier: Tier, queue: str) -> None: 

56 super().__init__(queue) 

57 self.tier = tier 

58 

59 @property 

60 def name(self) -> str: 

61 return f"tier={self.tier}" 

62 

63 def matches(self, draft: CaseDraft) -> bool: 

64 return draft.tier is self.tier 

65 

66 

67class ChannelRule(RoutingRule): 

68 def __init__(self, channel: Channel, queue: str) -> None: 

69 super().__init__(queue) 

70 self.channel = channel 

71 

72 @property 

73 def name(self) -> str: 

74 return f"channel={self.channel}" 

75 

76 def matches(self, draft: CaseDraft) -> bool: 

77 return draft.channel is self.channel 

78 

79 

80class KeywordRule(RoutingRule): 

81 """Matches whole words in the subject or description, ignoring case. 

82 

83 "refund" matches "Refund please" but not "refunded", so a rule only fires on the 

84 words it was written for. 

85 """ 

86 

87 def __init__(self, keywords: Iterable[str], queue: str) -> None: 

88 super().__init__(queue) 

89 cleaned = sorted({k.strip().lower() for k in keywords if k.strip()}) 

90 if not cleaned: 

91 raise ValueError("a keyword rule needs at least one keyword") 

92 self.keywords: tuple[str, ...] = tuple(cleaned) 

93 alternatives = "|".join(re.escape(k) for k in self.keywords) 

94 self._pattern = re.compile(rf"(?<!\w)(?:{alternatives})(?!\w)", re.IGNORECASE) 

95 

96 @property 

97 def name(self) -> str: 

98 return "keyword:" + ",".join(self.keywords) 

99 

100 def matches(self, draft: CaseDraft) -> bool: 

101 return bool(self._pattern.search(f"{draft.subject}\n{draft.description}")) 

102 

103 

104class AnyRule(RoutingRule): 

105 """Catch-all a tenant can put last to choose its own default queue.""" 

106 

107 @property 

108 def name(self) -> str: 

109 return "any" 

110 

111 def matches(self, draft: CaseDraft) -> bool: 

112 return True 

113 

114 

115class RuleSet: 

116 """An ordered list of rules. The first rule that matches decides the queue.""" 

117 

118 FALLBACK = RoutingDecision(queue="Triage", rule="fallback") 

119 

120 def __init__(self, rules: Sequence[RoutingRule]) -> None: 

121 self._rules = tuple(rules) 

122 

123 def route(self, draft: CaseDraft) -> RoutingDecision: 

124 for rule in self._rules: 

125 if rule.matches(draft): 

126 return RoutingDecision(queue=rule.queue, rule=rule.name) 

127 return self.FALLBACK 

128 

129 def __iter__(self) -> Iterator[RoutingRule]: 

130 return iter(self._rules) 

131 

132 def __len__(self) -> int: 

133 return len(self._rules) 

134 

135 

136_FACTORIES: dict[str, Callable[[str, str], RoutingRule]] = { 

137 "priority": lambda value, queue: MinimumPriorityRule(Priority(value.upper()), queue), 

138 "tier": lambda value, queue: TierRule(Tier(value.upper()), queue), 

139 "channel": lambda value, queue: ChannelRule(Channel(value.upper()), queue), 

140 "keyword": lambda value, queue: KeywordRule(value.split(","), queue), 

141 "any": lambda _value, queue: AnyRule(queue), 

142} 

143 

144RULE_KINDS = tuple(_FACTORIES) 

145 

146 

147def build_rule(kind: str, value: str, queue: str) -> RoutingRule: 

148 """Turn one stored rule row into a rule object. 

149 

150 Raises ValueError for an unknown kind or a value that does not fit the kind, so a bad 

151 row fails loudly instead of silently routing cases to the wrong queue. 

152 """ 

153 factory = _FACTORIES.get(kind) 

154 if factory is None: 

155 raise ValueError(f"unknown rule kind {kind!r}; expected one of {', '.join(RULE_KINDS)}") 

156 return factory(value.strip(), queue)