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

Refactorizing function to reduce its Cognitive Complexity

Browse Source
This commit is contained in:
Eduardo Rios
2026-09-15 14:35:22 -03:00
parent 9df151d29d
commit 86962c2cfc
1 changed files with 19 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
orchestrator/activities/slot_manager.py
View File

@@ -449,11 +449,8 @@ class SlotManager(SientiaMonitoring):
alert_type in receiver_group['contents']
and notification_id not in ignore_list
and key not in already_added_keys
and (not levels_list or notification.get('level') in levels_list)
and (
not ignore_models_id_list
or notification.get('model_id') not in ignore_models_id_list
)
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)
@@ -467,6 +464,23 @@ class SlotManager(SientiaMonitoring):
return receiver_groups
@staticmethod
def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list):
"""
Check if the notification's level is in the levels_list and its model_id is not in the ignore_models_id_list.
Args:
notification (dict): The notification to check.
levels_list (list, optional): List of levels to include. If None, all levels are included.
ignore_models_id_list (list, optional): List of model_ids to exclude. If None, no models are excluded.
Returns:
bool: True if the notification passes the checks, False otherwise.
"""
level_check = not levels_list or notification.get('level') in levels_list
model_id_check = not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list
return level_check and model_id_check
@activity.defn(name='store_notification_cache')
def store_notification_cache(self, input_data: dict[str, Any]) -> None:
"""