gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor: extract per-group filtering to reduce cognitive complexity

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-15 15:05:59 -03:00
parent 081298a883
commit b7325938f5
1 changed files with 67 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
orchestrator/activities/slot_manager.py
View File

@@ -417,54 +417,80 @@ class SlotManager(SientiaMonitoring):
for receiver_group in sending_configs: for receiver_group in sending_configs:
group_name = receiver_group['group_name'] group_name = receiver_group['group_name']
receiver_groups[group_name] = {**receiver_group, 'notifications': []} notifications = self._filter_group_notifications(
receiver_groups[group_name]['notifications'] = [] receiver_group, notification_package, notification_ttl
)
already_added_keys: list[str] = [] receiver_groups[group_name] = {**receiver_group, 'notifications': notifications}
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)
# Remove groups with no notifications # Remove groups with no notifications
receiver_groups = { return {
group_name: group group_name: group
for group_name, group in receiver_groups.items() for group_name, group in receiver_groups.items()
if group['notifications'] 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 @staticmethod
def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list): def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list):