refactor(forwarder): simplify code in retry and parsing modules

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: LorisFriedel

aws/logs_monitoring/retry/sqs_storage.py

@@ -53,12 +53,10 @@ def get_data(self, prefix):
                 msg_retry_prefix = self._get_message_attr(message, "retry_prefix")
                 msg_function_prefix = self._get_message_attr(message, "function_prefix")
 
-                matches = (
-                    msg_retry_prefix == str(prefix)
-                    and msg_function_prefix == self.function_prefix
-                )
-
-                if not matches:
+                if (
+                    msg_retry_prefix != str(prefix)
+                    or msg_function_prefix != self.function_prefix
+                ):
                     self._release_message(receipt_handle)
                     continue
 
@@ -123,9 +121,8 @@ def _release_message(self, receipt_handle):
     @staticmethod
     def _get_message_attr(message, attr_name):
         """Extract a string attribute value from an SQS message."""
-        return (
-            message.get("MessageAttributes", {}).get(attr_name, {}).get("StringValue")
-        )
+        attrs = message.get("MessageAttributes", {})
+        return attrs.get(attr_name, {}).get("StringValue")
 
     def _chunk_data(self, data):
         """Split a list of items into chunks that each fit under SQS_MAX_CHUNK_BYTES."""

aws/logs_monitoring/retry/storage.py

@@ -82,7 +82,7 @@ def _get_key_prefix(self, retry_prefix):
         return f"{DD_S3_RETRY_DIRNAME}/{self.function_prefix}/{str(retry_prefix)}/"
 
     def _serialize(self, data):
-        return bytes(json.dumps(data).encode("UTF-8"))
+        return json.dumps(data).encode("UTF-8")
 
     def _deserialize(self, data):
         return json.loads(data.decode("UTF-8"))

aws/logs_monitoring/steps/parsing.py

@@ -247,12 +247,6 @@ def normalize_events(events, metadata):
 
 
 def collect_and_count(events):
-    collected = []
-    counter = 0
-    for event in events:
-        counter += 1
-        collected.append(event)
-
-    send_event_metric("incoming_events", counter)
-
+    collected = list(events)
+    send_event_metric("incoming_events", len(collected))
     return collected