chore: cleanup exceptions

Author: heitorlessa

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -2,20 +2,16 @@
 Batch processing exceptions
 """
 import traceback
+from typing import Optional, Tuple
 
 
-class SQSBatchProcessingError(Exception):
-    """When at least one message within a batch could not be processed"""
-
+class BaseBatchProcessingError(Exception):
     def __init__(self, msg="", child_exceptions=()):
         super().__init__(msg)
         self.msg = msg
         self.child_exceptions = child_exceptions
 
-    # Overriding this method so we can output all child exception tracebacks when we raise this exception to prevent
-    # errors being lost. See https://github.com/awslabs/aws-lambda-powertools-python/issues/275
-    def __str__(self):
-        parent_exception_str = super(SQSBatchProcessingError, self).__str__()
+    def format_exceptions(self, parent_exception_str):
         exception_list = [f"{parent_exception_str}\n"]
         for exception in self.child_exceptions:
             extype, ex, tb = exception
@@ -25,22 +21,25 @@ def __str__(self):
         return "\n".join(exception_list)
 
 
-class BatchProcessingError(Exception):
-    """When batch messages could not be processed"""
+class SQSBatchProcessingError(BaseBatchProcessingError):
+    """When at least one message within a batch could not be processed"""
 
-    def __init__(self, msg="", child_exceptions=()):
-        super().__init__(msg)
-        self.msg = msg
-        self.child_exceptions = child_exceptions
+    def __init__(self, msg="", child_exceptions: Optional[Tuple[Exception]] = None):
+        super().__init__(msg, child_exceptions)
 
     # Overriding this method so we can output all child exception tracebacks when we raise this exception to prevent
     # errors being lost. See https://github.com/awslabs/aws-lambda-powertools-python/issues/275
     def __str__(self):
-        parent_exception_str = super(BatchProcessingError, self).__str__()
-        exception_list = [f"{parent_exception_str}\n"]
-        for exception in self.child_exceptions:
-            extype, ex, tb = exception
-            formatted = "".join(traceback.format_exception(extype, ex, tb))
-            exception_list.append(formatted)
+        parent_exception_str = super(SQSBatchProcessingError, self).__str__()
+        return self.format_exceptions(parent_exception_str)
 
-        return "\n".join(exception_list)
+
+class BatchProcessingError(BaseBatchProcessingError):
+    """When all batch records failed to be processed"""
+
+    def __init__(self, msg="", child_exceptions: Optional[Tuple[Exception]] = None):
+        super().__init__(msg, child_exceptions)
+
+    def __str__(self):
+        parent_exception_str = super(BatchProcessingError, self).__str__()
+        return self.format_exceptions(parent_exception_str)