refactor(python): enhance CloudWatch alarm scheduler with improved method organization

Refactored the CloudWatchAlarmScheduler class to improve code
clarity and maintainability. Added detailed docstrings for methods,
reorganized start and stop functionalities to utilize a shared
processing method, and introduced logging for actions taken on CloudWatch alarms.

Author: diodonfrost

package/scheduler/cloudwatch_handler.py

@@ -1,72 +1,76 @@
 """Cloudwatch alarm action scheduler."""
 
-from typing import Dict, List
+from typing import Dict, List, Optional
+import logging
 
 import boto3
-
 from botocore.exceptions import ClientError
 
 from .exceptions import cloudwatch_exception
 from .filter_resources_by_tags import FilterByTags
 
 
 class CloudWatchAlarmScheduler:
-    """Abstract Cloudwatch alarm scheduler in a class."""
+    """CloudWatch alarm scheduler for enabling/disabling alarm actions based on tags.
 
-    def __init__(self, region_name=None) -> None:
-        """Initialize Cloudwatch alarm scheduler."""
-        if region_name:
-            self.cloudwatch = boto3.client("cloudwatch", region_name=region_name)
-        else:
-            self.cloudwatch = boto3.client("cloudwatch")
-        self.tag_api = FilterByTags(region_name=region_name)
+    This class provides functionality to start (enable) or stop (disable) CloudWatch
+    alarm actions that match specific AWS resource tags.
+    """
 
-    def stop(self, aws_tags: List[Dict]) -> None:
-        """Aws Cloudwatch alarm disable function.
+    def __init__(self, region_name: Optional[str] = None) -> None:
+        """Initialize CloudWatch alarm scheduler with the specified AWS region.
 
-        Disable Cloudwatch alarm with defined tags.
+        Args:
+            region_name: Optional AWS region name. If not provided, the default
+                         region from AWS configuration will be used.
+        """
+        self.cloudwatch = (
+            boto3.client("cloudwatch", region_name=region_name)
+            if region_name
+            else boto3.client("cloudwatch")
+        )
+        self.tag_api = FilterByTags(region_name=region_name)
+        self.logger = logging.getLogger(__name__)
 
-        :param list[map] aws_tags:
-            Aws tags to use for filter resources.
-            For example:
-            [
-                {
-                    'Key': 'string',
-                    'Values': [
-                        'string',
-                    ]
-                }
-            ]
+    def _process_alarms(self, aws_tags: List[Dict], enable: bool) -> None:
+        """Process CloudWatch alarms by enabling or disabling them.
+
+        Args:
+            aws_tags: AWS tags to filter resources by
+            enable: True to enable alarms, False to disable alarms
         """
+        action = "enable" if enable else "disable"
+        method = (
+            self.cloudwatch.enable_alarm_actions
+            if enable
+            else self.cloudwatch.disable_alarm_actions
+        )
+        action_present = "Enabling" if enable else "Disabling"
+        action_past = "Enabled" if enable else "Disabled"
+
         for alarm_arn in self.tag_api.get_resources("cloudwatch:alarm", aws_tags):
             alarm_name = alarm_arn.split(":")[-1]
             try:
-                self.cloudwatch.disable_alarm_actions(AlarmNames=[alarm_name])
-                print(f"Disable Cloudwatch alarm {alarm_name}")
+                method(AlarmNames=[alarm_name])
+                self.logger.info(f"{action_past} CloudWatch alarm {alarm_name}")
+                print(f"{action_past} CloudWatch alarm {alarm_name}")
             except ClientError as exc:
                 cloudwatch_exception("cloudwatch alarm", alarm_name, exc)
 
-    def start(self, aws_tags: List[Dict]) -> None:
-        """Aws Cloudwatch alarm enable function.
+    def stop(self, aws_tags: List[Dict]) -> None:
+        """Disable CloudWatch alarm actions for resources with the specified tags.
 
-        Enable Cloudwatch alarm with defined tags.
+        Args:
+            aws_tags: AWS tags to filter resources by.
+                      Format: [{'Key': 'tag_key', 'Values': ['tag_value', ...]}]
+        """
+        self._process_alarms(aws_tags, enable=False)
+
+    def start(self, aws_tags: List[Dict]) -> None:
+        """Enable CloudWatch alarm actions for resources with the specified tags.
 
-        :param list[map] aws_tags:
-            Aws tags to use for filter resources.
-            For example:
-            [
-                {
-                    'Key': 'string',
-                    'Values': [
-                        'string',
-                    ]
-                }
-            ]
+        Args:
+            aws_tags: AWS tags to filter resources by.
+                      Format: [{'Key': 'tag_key', 'Values': ['tag_value', ...]}]
         """
-        for alarm_arn in self.tag_api.get_resources("cloudwatch:alarm", aws_tags):
-            alarm_name = alarm_arn.split(":")[-1]
-            try:
-                self.cloudwatch.enable_alarm_actions(AlarmNames=[alarm_name])
-                print(f"Enable Cloudwatch alarm {alarm_name}")
-            except ClientError as exc:
-                cloudwatch_exception("cloudwatch alarm", alarm_name, exc)
+        self._process_alarms(aws_tags, enable=True)