Coverage for extras/conditions.py: 18%
188 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import re
3from django.utils.translation import gettext as _
5__all__ = (
6 'AbsentData',
7 'Condition',
8 'ConditionSet',
9 'InvalidCondition',
10)
12AND = 'and'
13OR = 'or'
15# Prefix identifying a condition attribute that reads an event's pre- or post-change snapshot directly, e.g.
16# 'snapshots.prechange.status'.
17SNAPSHOT_PREFIX = 'snapshots.'
19# Maps each snapshot to its counterpart
20OPPOSITE_SNAPSHOT = {
21 'prechange': 'postchange',
22 'postchange': 'prechange',
23}
25# Sentinel for a snapshot attribute that could not be resolved (missing key or null snapshot)
26_MISSING = object()
29class AbsentData(dict):
30 """
31 An empty dict standing in for an event payload which cannot be evaluated: one which is
32 absent (a job which recorded no data) or unusable (a payload which is not a dict at all).
33 """
34 def copy(self):
35 # dict.copy() would return a plain dict, silently discarding the marker.
36 return AbsentData(self)
39def walk_path(obj, keys, empty_list_is_absent=False):
40 """
41 Walk a sequence of keys through obj, returning _MISSING if a key is absent or null along the way.
43 Raises TypeError if the path descends into a value which cannot be indexed by key (e.g. a
44 REST API-style 'status.value' applied to a snapshot, where status is the raw string
45 "active"). Walkability follows from the value's type: an empty string is as unwalkable as any
46 other scalar, not an absent key.
47 """
48 for key in keys:
49 if obj is None:
50 return _MISSING
51 if isinstance(obj, list):
52 if not obj and empty_list_is_absent:
53 # An empty list yields no evidence either way
54 return _MISSING
55 values = []
56 for item in obj:
57 if item is None:
58 return _MISSING
59 if not isinstance(item, dict):
60 raise TypeError(f"cannot resolve '{key}' within {type(item).__name__}")
61 if key not in item:
62 return _MISSING
63 values.append(item[key])
64 obj = values
65 elif isinstance(obj, dict):
66 if key not in obj:
67 return _MISSING
68 obj = obj[key]
69 else:
70 raise TypeError(f"cannot resolve '{key}' within {type(obj).__name__}")
71 return obj
74def is_ruleset(data):
75 """
76 Determine whether the given dictionary looks like a rule set.
77 """
78 return type(data) is dict and len(data) == 1 and list(data.keys())[0] in (AND, OR)
81class InvalidCondition(Exception):
82 pass
85class Condition:
86 """
87 An individual conditional rule that evaluates a single attribute and its value.
89 :param attr: The name of the attribute being evaluated
90 :param value: The value being compared (not used by snapshot operators)
91 :param op: The logical operation to use when evaluating the value (default: 'eq')
92 :param negate: Invert the result of evaluation
93 """
94 EQ = 'eq'
95 GT = 'gt'
96 GTE = 'gte'
97 LT = 'lt'
98 LTE = 'lte'
99 IN = 'in'
100 CONTAINS = 'contains'
101 REGEX = 'regex'
102 CHANGED = 'changed'
103 UNCHANGED = 'unchanged'
105 OPERATORS = (
106 EQ, GT, GTE, LT, LTE, IN, CONTAINS, REGEX, CHANGED, UNCHANGED
107 )
109 # Operators that compare pre/post snapshots and do not accept a value.
110 SNAPSHOT_OPERATORS = (CHANGED, UNCHANGED)
112 TYPES = {
113 str: (EQ, CONTAINS, REGEX),
114 bool: (EQ, CONTAINS),
115 int: (EQ, GT, GTE, LT, LTE, CONTAINS),
116 float: (EQ, GT, GTE, LT, LTE, CONTAINS),
117 list: (EQ, IN, CONTAINS),
118 type(None): (EQ,)
119 }
121 def __init__(self, attr, value=_MISSING, op=EQ, negate=False):
122 if op not in self.OPERATORS:
123 raise ValueError(_("Unknown operator: {op}. Must be one of: {operators}").format(
124 op=op, operators=', '.join(self.OPERATORS)
125 ))
127 if op in self.SNAPSHOT_OPERATORS:
128 if value is not _MISSING:
129 raise ValueError(_(
130 "The '{op}' operator compares snapshots and does not accept a value."
131 ).format(op=op))
132 if attr.startswith(SNAPSHOT_PREFIX):
133 raise ValueError(_(
134 "The '{op}' operator resolves '{attr}' within each snapshot dict, not the "
135 "top-level condition context. Use the bare attribute name (e.g. 'status') "
136 "rather than a snapshot path (e.g. 'snapshots.prechange.status'), which is "
137 "only valid with standard operators."
138 ).format(op=op, attr=attr))
139 self.value = _MISSING
140 else:
141 if value is _MISSING:
142 raise ValueError(_("A value is required for the '{op}' operator.").format(op=op))
143 if type(value) not in self.TYPES:
144 raise ValueError(_("Unsupported value type: {value}").format(value=type(value)))
145 if op not in self.TYPES[type(value)]:
146 raise ValueError(_("Invalid type for {op} operation: {value}").format(op=op, value=type(value)))
147 self.value = value
149 self.attr = attr
150 self.op = op
151 self.eval_func = getattr(self, f'eval_{op}')
152 self.negate = negate
154 def _resolve_attr(self, data):
155 """
156 Walk self.attr as a dotted key path through data. Raises InvalidCondition on
157 missing keys, or when an intermediate value can't be indexed by key (e.g. a
158 REST API-style path like 'status.value' applied to a raw snapshot value).
159 """
160 try:
161 value = walk_path(data, self.attr.split('.'))
162 except TypeError as e:
163 raise InvalidCondition(f"Invalid key path: {self.attr} ({e})")
164 if value is _MISSING:
165 raise InvalidCondition(f"Invalid key path: {self.attr}")
166 return value
168 def _references_absent_payload(self, data):
169 """
170 Return True if self.attr references an attribute of a payload which is absent
171 altogether (AbsentData), as opposed to one which is present but lacks the attribute.
172 """
173 return isinstance(data, AbsentData) and self.attr.split('.')[0] not in data
175 def _references_absent_snapshot(self, data):
176 """
177 Return True if self.attr is a direct snapshot path (snapshots.prechange.* or
178 snapshots.postchange.*) whose snapshot is null and whose remaining path the opposite
179 snapshot resolves. Create events have no prechange snapshot, delete events no
180 postchange snapshot.
182 Unlike an absent payload, such a reference resolves to null: the snapshot's absence is
183 itself meaningful (the object did not exist before, or does not after), and validating
184 the path below shows the reference to describe the data.
185 """
186 if not self.attr.startswith(SNAPSHOT_PREFIX):
187 return False
188 snapshots = data.get('snapshots') if isinstance(data, dict) else None
189 if type(snapshots) is not dict:
190 return False
191 which, _sep, remainder = self.attr[len(SNAPSHOT_PREFIX):].partition('.')
192 if which not in OPPOSITE_SNAPSHOT:
193 # Anything other than prechange or postchange names no snapshot the event could have
194 # recorded, so the path does not describe the data
195 return False
196 if which not in snapshots or snapshots[which] is not None:
197 return False
199 # The referenced snapshot is null, which excuses only data the event would otherwise
200 # have carried, never a path which does not describe the data. Validate the remainder
201 # against the opposite snapshot so that such a path fails closed here exactly as it does
202 # when both snapshots are present; otherwise a typo would resolve to null and fire the
203 # rule on every create or delete, with nothing logged.
204 other = snapshots.get(OPPOSITE_SNAPSHOT[which])
205 if remainder and other is not None:
206 try:
207 value = walk_path(other, remainder.split('.'), empty_list_is_absent=True)
208 except TypeError:
209 return False
210 if value is _MISSING:
211 return False
213 # Nothing to validate against: with the opposite snapshot absent too, the event carries
214 # no data anywhere for the path to be checked. Testing for the absent snapshot itself
215 # (snapshots.prechange, no remainder) lands here too.
216 return True
218 def _resolve_snapshot_attrs(self, snapshots):
219 """
220 Walk self.attr through the prechange and postchange snapshots, returning the two
221 resolved values, with _MISSING for a snapshot which is absent, lacks the attribute, or
222 cannot be walked by the path.
224 Raises InvalidCondition if the attribute resolves in neither snapshot, leaving nothing
225 to compare: a misspelling, an unwalkable path, or an event which recorded no snapshots.
226 The unresolved state must be reported rather than compared, since any boolean it
227 returned would become a match under negate.
229 A path which resolves in only one snapshot describes a real difference between them (a
230 JSON attribute whose value changed shape, say), so the unresolved side counts as missing
231 and the comparison proceeds: raising would report as unchanged an attribute which
232 demonstrably changed. Only a snapshot yielding a value excuses the other side; one
233 resolving to nothing is no evidence that the path describes the data.
234 """
235 keys = self.attr.split('.')
236 values = []
237 errors = []
238 available = False
239 resolved = False
241 for which in ('prechange', 'postchange'):
242 snapshot = snapshots.get(which)
243 if snapshot is None:
244 # Absent snapshot (normal for create and delete events): nothing to resolve
245 values.append(_MISSING)
246 continue
247 available = True
248 try:
249 value = walk_path(snapshot, keys, empty_list_is_absent=True)
250 except TypeError as e:
251 values.append(_MISSING)
252 errors.append(e)
253 else:
254 values.append(value)
255 resolved = resolved or value is not _MISSING
257 if not available:
258 # Neither snapshot was recorded, so the attribute itself is not in question
259 raise InvalidCondition(
260 f"No snapshot data available for '{self.op}' operator: {self.attr}. "
261 f"Snapshot operators are only meaningful on update and delete events."
262 )
263 if not resolved:
264 reason = f" ({errors[0]})" if errors else ""
265 raise InvalidCondition(
266 f"Invalid key path for '{self.op}' operator: {self.attr}{reason}. The attribute resolves in neither "
267 f"snapshot. Note that snapshots store raw field values, so choice fields have no '.value' suffix."
268 )
270 return values
272 def eval(self, data):
273 """
274 Evaluate the provided data to determine whether it matches the condition.
275 """
276 if self.op in self.SNAPSHOT_OPERATORS:
277 snapshots = data.get('snapshots') if isinstance(data, dict) else None
278 if type(snapshots) is not dict:
279 raise InvalidCondition(
280 f"No snapshot data available for '{self.op}' operator. "
281 f"Snapshot operators are only meaningful on update and delete events."
282 )
283 result = self.eval_func(snapshots)
284 return not result if self.negate else result
286 if self._references_absent_payload(data):
287 # No payload to evaluate, so the condition cannot be satisfied. Negation is not
288 # applied: it inverts the result of a comparison, and none took place - inverting
289 # would fire the rule on an event which carried nothing to match against. Nor is
290 # this an invalid condition: a job which records no data is routine, and logging it
291 # per rule per event would bury the malformed conditions worth acting on.
292 return False
294 absent = self._references_absent_snapshot(data)
295 value = None if absent else self._resolve_attr(data)
296 try:
297 result = self.eval_func(value)
298 except TypeError as e:
299 if not absent:
300 raise InvalidCondition(f"Invalid data type at '{self.attr}' for '{self.op}' evaluation: {e}")
301 # An absent snapshot resolves to null, which satisfies only a comparison against
302 # null: contains, regex and the numeric comparisons raise TypeError on None. That is
303 # a non-match, not a malformed condition, so report False (subject to negation
304 # below) rather than aborting the condition set.
305 result = False
307 if self.negate:
308 return not result
309 return result
311 # Equivalency
313 def eval_eq(self, value):
314 return value == self.value
316 def eval_neq(self, value):
317 return value != self.value
319 # Numeric comparisons
321 def eval_gt(self, value):
322 return value > self.value
324 def eval_gte(self, value):
325 return value >= self.value
327 def eval_lt(self, value):
328 return value < self.value
330 def eval_lte(self, value):
331 return value <= self.value
333 # Membership
335 def eval_in(self, value):
336 return value in self.value
338 def eval_contains(self, value):
339 return self.value in value
341 # Regular expressions
343 def eval_regex(self, value):
344 return re.match(self.value, value) is not None
346 # Snapshot comparison operators
348 def eval_changed(self, snapshots):
349 pre, post = self._resolve_snapshot_attrs(snapshots)
350 return pre != post
352 def eval_unchanged(self, snapshots):
353 pre, post = self._resolve_snapshot_attrs(snapshots)
354 return pre == post
357class ConditionSet:
358 """
359 A set of one or more Condition to be evaluated per the prescribed logic (AND or OR). Example:
361 {"and": [
362 {"attr": "foo", "op": "eq", "value": 1},
363 {"attr": "bar", "op": "eq", "value": 2, "negate": true}
364 ]}
366 :param ruleset: A dictionary mapping a logical operator to a list of conditional rules
367 """
368 def __init__(self, ruleset):
369 if type(ruleset) is not dict:
370 raise ValueError(_("Ruleset must be a dictionary, not {ruleset}.").format(ruleset=type(ruleset)))
372 if len(ruleset) == 1:
373 self.logic = (list(ruleset.keys())[0]).lower()
374 if self.logic not in (AND, OR):
375 raise ValueError(_("Invalid logic type: must be 'AND' or 'OR'. Please check documentation."))
377 # Compile the set of Conditions
378 self.conditions = [
379 ConditionSet(rule) if is_ruleset(rule) else Condition(**rule)
380 for rule in ruleset[self.logic]
381 ]
382 else:
383 try:
384 self.logic = None
385 self.conditions = [Condition(**ruleset)]
386 except TypeError:
387 raise ValueError(_("Incorrect key(s) informed. Please check documentation."))
389 def eval(self, data):
390 """
391 Evaluate the provided data to determine whether it matches this set of conditions.
392 """
393 func = any if self.logic == 'or' else all
394 return func(d.eval(data) for d in self.conditions)