fix: satisfy batch logger mypy checks

Author: Sujit

aws_lambda_powertools/utilities/batch/base.py

@@ -14,7 +14,7 @@
 import sys
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Tuple, Union, overload
+from typing import TYPE_CHECKING, Any, Tuple, Union, cast, overload
 
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.utilities.batch.exceptions import (
@@ -35,6 +35,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Callable
+    from types import TracebackType
 
     from aws_lambda_powertools.logging import Logger
     from aws_lambda_powertools.utilities.batch.types import (
@@ -240,10 +241,19 @@ def failure_handler(self, record, exception: ExceptionInfo) -> FailureResponse:
         entry = ("fail", exception_string, record)
         logger.debug(f"Record processing exception: {exception_string}")
 
-        if getattr(self, "logger", None) and exception[2] is not None:
-            self.logger.warning(
+        # Log with full traceback when a customer-provided logger is present
+        # and the exception carries a real traceback (e.g. not a synthetic FIFO circuit-breaker)
+        batch_logger = self.logger
+        if batch_logger is not None and exception[2] is not None:
+            # ExceptionInfo allows None on every slot, but logging.warning's exc_info
+            # requires a fully populated tuple. We already excluded synthetic exceptions
+            # (no traceback) above, so the type and value are guaranteed to be set.
+            assert exception[0] is not None
+            assert exception[1] is not None
+            exc_info = cast("tuple[type[BaseException], BaseException, TracebackType]", exception)
+            batch_logger.warning(
                 "Record processing exception; skipping this record",
-                exc_info=exception,
+                exc_info=exc_info,
             )
 
         self.exceptions.append(exception)

tests/functional/batch/required_dependencies/test_utilities_batch.py

@@ -865,6 +865,7 @@ async def simple_async_handler(record: SQSRecord):
 
 def test_batch_processor_logs_exception_with_injected_logger(sqs_event_factory, caplog):
     import logging
+
     from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, process_partial_response
 
     fail_record = sqs_event_factory("fail")
@@ -894,6 +895,7 @@ def handler(record):
 
 def test_batch_processor_does_not_log_without_injected_logger(sqs_event_factory, caplog):
     import logging
+
     from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, process_partial_response
 
     fail_record = sqs_event_factory("fail")
@@ -916,6 +918,7 @@ def handler(record):
 
 def test_sqs_fifo_circuit_breaker_does_not_log(sqs_event_fifo_factory, caplog):
     import logging
+
     from aws_lambda_powertools.utilities.batch import SqsFifoPartialProcessor, process_partial_response
 
     failing_record = sqs_event_fifo_factory("fail", "group-1")