Move free functions into PushRuleEvaluatorForEvent. (#12677)

* Move `_condition_checker` into `PushRuleEvaluatorForEvent`.
* Move the condition cache into `PushRuleEvaluatorForEvent`.
* Improve docstrings.
* Inline a method which is only called once.
This commit is contained in:
Patrick Cloke 2022-05-10 07:54:30 -04:00 committed by GitHub
parent 02cdace707
commit b44fbdffa4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 69 additions and 34 deletions

1
changelog.d/12677.misc Normal file
View File

@ -0,0 +1 @@
Refactor functions to on `PushRuleEvaluatorForEvent`.

View File

@ -208,8 +208,6 @@ class BulkPushRuleEvaluator:
event, len(room_members), sender_power_level, power_levels
)
condition_cache: Dict[str, bool] = {}
# If the event is not a state event check if any users ignore the sender.
if not event.is_state():
ignorers = await self.store.ignored_by(event.sender)
@ -247,8 +245,8 @@ class BulkPushRuleEvaluator:
if "enabled" in rule and not rule["enabled"]:
continue
matches = _condition_checker(
evaluator, rule["conditions"], uid, display_name, condition_cache
matches = evaluator.check_conditions(
rule["conditions"], uid, display_name
)
if matches:
actions = [x for x in rule["actions"] if x != "dont_notify"]
@ -267,32 +265,6 @@ class BulkPushRuleEvaluator:
)
def _condition_checker(
evaluator: PushRuleEvaluatorForEvent,
conditions: List[dict],
uid: str,
display_name: Optional[str],
cache: Dict[str, bool],
) -> bool:
for cond in conditions:
_cache_key = cond.get("_cache_key", None)
if _cache_key:
res = cache.get(_cache_key, None)
if res is False:
return False
elif res is True:
continue
res = evaluator.matches(cond, uid, display_name)
if _cache_key:
cache[_cache_key] = bool(res)
if not res:
return False
return True
MemberMap = Dict[str, Optional[EventIdMembership]]
Rule = Dict[str, dict]
RulesByUser = Dict[str, List[Rule]]

View File

@ -129,9 +129,55 @@ class PushRuleEvaluatorForEvent:
# Maps strings of e.g. 'content.body' -> event["content"]["body"]
self._value_cache = _flatten_dict(event)
# Maps cache keys to final values.
self._condition_cache: Dict[str, bool] = {}
def check_conditions(
self, conditions: List[dict], uid: str, display_name: Optional[str]
) -> bool:
"""
Returns true if a user's conditions/user ID/display name match the event.
Args:
conditions: The user's conditions to match.
uid: The user's MXID.
display_name: The display name.
Returns:
True if all conditions match the event, False otherwise.
"""
for cond in conditions:
_cache_key = cond.get("_cache_key", None)
if _cache_key:
res = self._condition_cache.get(_cache_key, None)
if res is False:
return False
elif res is True:
continue
res = self.matches(cond, uid, display_name)
if _cache_key:
self._condition_cache[_cache_key] = bool(res)
if not res:
return False
return True
def matches(
self, condition: Dict[str, Any], user_id: str, display_name: Optional[str]
) -> bool:
"""
Returns true if a user's condition/user ID/display name match the event.
Args:
condition: The user's condition to match.
uid: The user's MXID.
display_name: The display name, or None if there is not one.
Returns:
True if the condition matches the event, False otherwise.
"""
if condition["kind"] == "event_match":
return self._event_match(condition, user_id)
elif condition["kind"] == "contains_display_name":
@ -146,6 +192,16 @@ class PushRuleEvaluatorForEvent:
return True
def _event_match(self, condition: dict, user_id: str) -> bool:
"""
Check an "event_match" push rule condition.
Args:
condition: The "event_match" push rule condition to match.
user_id: The user's MXID.
Returns:
True if the condition matches the event, False otherwise.
"""
pattern = condition.get("pattern", None)
if not pattern:
@ -167,13 +223,22 @@ class PushRuleEvaluatorForEvent:
return _glob_matches(pattern, body, word_boundary=True)
else:
haystack = self._get_value(condition["key"])
haystack = self._value_cache.get(condition["key"], None)
if haystack is None:
return False
return _glob_matches(pattern, haystack)
def _contains_display_name(self, display_name: Optional[str]) -> bool:
"""
Check an "event_match" push rule condition.
Args:
display_name: The display name, or None if there is not one.
Returns:
True if the display name is found in the event body, False otherwise.
"""
if not display_name:
return False
@ -191,9 +256,6 @@ class PushRuleEvaluatorForEvent:
return bool(r.search(body))
def _get_value(self, dotted_key: str) -> Optional[str]:
return self._value_cache.get(dotted_key, None)
# Caches (string, is_glob, word_boundary) -> regex for push. See _glob_matches
regex_cache: LruCache[Tuple[str, bool, bool], Pattern] = LruCache(