From b7325938f5dcc2dbacd76947503025acbe819a49 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 15:05:59 -0300 Subject: [PATCH] refactor: extract per-group filtering to reduce cognitive complexity filter_notification_alerts tripped Sonar's cognitive-complexity gate (16 > 15 allowed) from its nested loop/if structure. Split into _filter_group_notifications (per-group notification selection) and _determine_alert_type (Redis TTL classification), leaving the public method as a simple loop over receiver groups. Behavior and call order are unchanged -- verified against the existing filter_notification_alerts unit tests by inspection (sientia_do isn't installable in this local environment, so the real suite can't run here; CI will confirm). Co-Authored-By: Claude Sonnet 5 --- orchestrator/activities/slot_manager.py | 108 +++++++++++++++--------- 1 file changed, 67 insertions(+), 41 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 395f094..7182551 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -417,54 +417,80 @@ class SlotManager(SientiaMonitoring): for receiver_group in sending_configs: group_name = receiver_group['group_name'] - receiver_groups[group_name] = {**receiver_group, 'notifications': []} - receiver_groups[group_name]['notifications'] = [] - - already_added_keys: list[str] = [] - - ignore_list = receiver_group.get('ignore', []) - levels_list = receiver_group.get('levels') - ignore_models_id_list = receiver_group.get('ignore_models_id') or [] - - for notification in notification_package: - alert_type = 'do_nothing' - notification_id = notification['notification_id'] - # Check if notification was recently sent - key = f'{notification["trigger"]}:{notification_id}' - - last_sent = self.redis_repository.get(key) - - if last_sent is None: - alert_type = 'core_alerts' - - else: - last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ) - - # Check if "notification_ttl" seconds has passed since last sent - if (now() - last_sent) > timedelta(seconds=notification_ttl): - alert_type = 'persistent_alerts' - - # Check if this group must be notified - if ( - alert_type in receiver_group['contents'] - and self._check_keys_and_notif_id( - notification_id, ignore_list, key, already_added_keys - ) - and self._check_levels_and_ignore_models_id( - notification, levels_list, ignore_models_id_list - ) - ): - receiver_groups[group_name]['notifications'].append(notification) - already_added_keys.append(key) + notifications = self._filter_group_notifications( + receiver_group, notification_package, notification_ttl + ) + receiver_groups[group_name] = {**receiver_group, 'notifications': notifications} # Remove groups with no notifications - receiver_groups = { + return { group_name: group for group_name, group in receiver_groups.items() if group['notifications'] } - return receiver_groups + def _filter_group_notifications( + self, + receiver_group: dict[str, Any], + notification_package: list[dict[str, Any]], + notification_ttl: int, + ) -> list[dict[str, Any]]: + """ + Select the notifications this receiver group should get. + + Args: + receiver_group (dict[str, Any]): The receiver group configuration. + notification_package (list[dict[str, Any]]): Candidate notifications. + notification_ttl (int): Seconds before considering a notification persistent. + + Returns: + list[dict[str, Any]]: Notifications matching this group's filters. + """ + already_added_keys: list[str] = [] + ignore_list = receiver_group.get('ignore', []) + levels_list = receiver_group.get('levels') + ignore_models_id_list = receiver_group.get('ignore_models_id') or [] + + notifications = [] + for notification in notification_package: + notification_id = notification['notification_id'] + key = f'{notification["trigger"]}:{notification_id}' + alert_type = self._determine_alert_type(key, notification_ttl) + + if ( + alert_type in receiver_group['contents'] + and self._check_keys_and_notif_id( + notification_id, ignore_list, key, already_added_keys + ) + and self._check_levels_and_ignore_models_id( + notification, levels_list, ignore_models_id_list + ) + ): + notifications.append(notification) + already_added_keys.append(key) + + return notifications + + def _determine_alert_type(self, key: str, notification_ttl: int) -> str: + """ + Classify a notification as new, persistent, or not-yet-due. + + Args: + key (str): The Redis cache key for this notification. + notification_ttl (int): Seconds before considering a notification persistent. + + Returns: + str: 'core_alerts', 'persistent_alerts', or 'do_nothing'. + """ + last_sent = self.redis_repository.get(key) + if last_sent is None: + return 'core_alerts' + + last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ) + if (now() - last_sent) > timedelta(seconds=notification_ttl): + return 'persistent_alerts' + + return 'do_nothing' @staticmethod def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list):