chore: fix child exceptions type; remove casting

Author: heitorlessa

aws_lambda_powertools/utilities/batch/base.py

@@ -31,7 +31,6 @@ class EventType(Enum):
 #
 has_pydantic = "pydantic" in sys.modules
 ExceptionInfo = Tuple[Type[BaseException], BaseException, TracebackType]
-OptExcInfo = Union[ExceptionInfo, Tuple[None, None, None]]
 
 # For IntelliSense and Mypy to work, we need to account for possible SQS, Kinesis and DynamoDB subclasses
 # We need them as subclasses as we must access their message ID or sequence number metadata via dot notation
@@ -61,7 +60,7 @@ class BasePartialProcessor(ABC):
     def __init__(self):
         self.success_messages: List[BatchEventTypes] = []
         self.fail_messages: List[BatchEventTypes] = []
-        self.exceptions: List = []
+        self.exceptions: List[ExceptionInfo] = []
 
     @abstractmethod
     def _prepare(self):
@@ -132,15 +131,15 @@ def success_handler(self, record, result: Any) -> SuccessResponse:
         self.success_messages.append(record)
         return entry
 
-    def failure_handler(self, record, exception: OptExcInfo) -> FailureResponse:
+    def failure_handler(self, record, exception: ExceptionInfo) -> FailureResponse:
         """
         Keeps track of batch records that failed processing
 
         Parameters
         ----------
         record: Any
             record that failed processing
-        exception: OptExcInfo
+        exception: ExceptionInfo
             Exception information containing type, value, and traceback (sys.exc_info())
 
         Returns
@@ -384,7 +383,7 @@ def _clean(self):
             raise BatchProcessingError(
                 msg=f"All records failed processing. {len(self.exceptions)} individual errors logged"
                 f"separately below.",
-                child_exceptions=tuple(self.exceptions),
+                child_exceptions=self.exceptions,
             )
 
         messages = self._get_messages_to_report()

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -2,11 +2,13 @@
 Batch processing exceptions
 """
 import traceback
-from typing import Optional, Tuple
+from typing import List, Optional
+
+from aws_lambda_powertools.utilities.batch import ExceptionInfo
 
 
 class BaseBatchProcessingError(Exception):
-    def __init__(self, msg="", child_exceptions=()):
+    def __init__(self, msg="", child_exceptions: Optional[List[ExceptionInfo]] = None):
         super().__init__(msg)
         self.msg = msg
         self.child_exceptions = child_exceptions
@@ -24,7 +26,7 @@ def format_exceptions(self, parent_exception_str):
 class SQSBatchProcessingError(BaseBatchProcessingError):
     """When at least one message within a batch could not be processed"""
 
-    def __init__(self, msg="", child_exceptions: Optional[Tuple[Exception]] = None):
+    def __init__(self, msg="", child_exceptions: Optional[List[ExceptionInfo]] = None):
         super().__init__(msg, child_exceptions)
 
     # Overriding this method so we can output all child exception tracebacks when we raise this exception to prevent
@@ -37,7 +39,7 @@ def __str__(self):
 class BatchProcessingError(BaseBatchProcessingError):
     """When all batch records failed to be processed"""
 
-    def __init__(self, msg="", child_exceptions: Optional[Tuple[Exception]] = None):
+    def __init__(self, msg="", child_exceptions: Optional[List[ExceptionInfo]] = None):
         super().__init__(msg, child_exceptions)
 
     def __str__(self):

tests/functional/test_utilities_batch.py

@@ -832,5 +832,7 @@ def lambda_handler(event, context):
         return processor.response()
 
     # WHEN/THEN
-    with pytest.raises(BatchProcessingError):
+    with pytest.raises(BatchProcessingError) as e:
         lambda_handler(event, {})
+        ret = str(e)
+        assert ret is not None