From a87e0cbd325583ca50933d3a82469124fe570f35 Mon Sep 17 00:00:00 2001 From: AI Bot Date: Tue, 26 May 2026 00:39:55 +0530 Subject: [PATCH 1/3] chore: fix minor typos and grammar in docs and comments --- aws_lambda_powertools/utilities/parameters/ssm.py | 2 +- docs/utilities/parser.md | 4 ++-- .../event_handler/_pydantic/test_openapi_responses.py | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/aws_lambda_powertools/utilities/parameters/ssm.py b/aws_lambda_powertools/utilities/parameters/ssm.py index fde8d980494..7c2b5e1017e 100644 --- a/aws_lambda_powertools/utilities/parameters/ssm.py +++ b/aws_lambda_powertools/utilities/parameters/ssm.py @@ -504,7 +504,7 @@ def get_parameters_by_name( # NOTE: We need to find out whether all parameters must be decrypted or not to know which API to use ## Logic: ## - ## GetParameters API -> When decrypt is used for all parameters in the the batch + ## GetParameters API -> When decrypt is used for all parameters in the batch ## GetParameter API -> When decrypt is used for one or more in the batch if len(decrypt_params) != len(parameters): diff --git a/docs/utilities/parser.md b/docs/utilities/parser.md index b4c7ad749a9..0a142caa18b 100644 --- a/docs/utilities/parser.md +++ b/docs/utilities/parser.md @@ -211,8 +211,8 @@ You can use pre-built envelopes provided by the Parser to extract and parse spec | **EventBridgeEnvelope** | 1. Parses data using `EventBridgeModel`. ``2. Parses `detail` key using your model`` and returns it. | `Model` | | **SqsEnvelope** | 1. Parses data using `SqsModel`. ``2. Parses records in `body` key using your model`` and return them in a list. | `List[Model]` | | **CloudWatchLogsEnvelope** | 1. Parses data using `CloudwatchLogsModel` which will base64 decode and decompress it. ``2. Parses records in `message` key using your model`` and return them in a list. | `List[Model]` | -| **KinesisDataStreamEnvelope** | 1. Parses data using `KinesisDataStreamModel` which will base64 decode it. ``2. Parses records in in `Records` key using your model`` and returns them in a list. | `List[Model]` | -| **KinesisFirehoseEnvelope** | 1. Parses data using `KinesisFirehoseModel` which will base64 decode it. ``2. Parses records in in` Records` key using your model`` and returns them in a list. | `List[Model]` | +| **KinesisDataStreamEnvelope** | 1. Parses data using `KinesisDataStreamModel` which will base64 decode it. ``2. Parses records in the `Records` key using your model`` and returns them in a list. | `List[Model]` | +| **KinesisFirehoseEnvelope** | 1. Parses data using `KinesisFirehoseModel` which will base64 decode it. ``2. Parses records in the `Records` key using your model`` and returns them in a list. | `List[Model]` | | **SnsEnvelope** | 1. Parses data using `SnsModel`. ``2. Parses records in `body` key using your model`` and return them in a list. | `List[Model]` | | **SnsSqsEnvelope** | 1. Parses data using `SqsModel`. `` 2. Parses SNS records in `body` key using `SnsNotificationModel`. `` 3. Parses data in `Message` key using your model and return them in a list. | `List[Model]` | | **ApiGatewayV2Envelope** | 1. Parses data using `APIGatewayProxyEventV2Model`. ``2. Parses `body` key using your model`` and returns it. | `Model` | diff --git a/tests/functional/event_handler/_pydantic/test_openapi_responses.py b/tests/functional/event_handler/_pydantic/test_openapi_responses.py index 785d2b8416c..dccfae3e28d 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_responses.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_responses.py @@ -232,7 +232,7 @@ def handler(): schema = app.get_openapi_schema() responses = schema.paths["/"].get.responses - # THE the schema should include a 200 successful response + # The schema should include a 200 successful response # but not a 422 validation error response since validation is disabled assert 200 in responses.keys() assert responses[200].description == "Successful Response" From 3090e4d742b9ed75ff7e029caa86ef53d2d9006f Mon Sep 17 00:00:00 2001 From: AI Bot Date: Sun, 14 Jun 2026 09:25:42 +0530 Subject: [PATCH 2/3] fix(batch): add optional logger injection for BatchProcessors (#7553) --- aws_lambda_powertools/utilities/batch/base.py | 16 +++- .../batch/sqs_fifo_partial_processor.py | 12 ++- .../test_utilities_batch.py | 82 +++++++++++++++++++ 3 files changed, 106 insertions(+), 4 deletions(-) diff --git a/aws_lambda_powertools/utilities/batch/base.py b/aws_lambda_powertools/utilities/batch/base.py index 4cb1337cafd..f51b17db6f5 100644 --- a/aws_lambda_powertools/utilities/batch/base.py +++ b/aws_lambda_powertools/utilities/batch/base.py @@ -36,6 +36,7 @@ if TYPE_CHECKING: from collections.abc import Callable + from aws_lambda_powertools.logging import Logger from aws_lambda_powertools.utilities.batch.types import ( PartialItemFailureResponse, PartialItemFailures, @@ -68,10 +69,11 @@ class BasePartialProcessor(ABC): lambda_context: LambdaContext - def __init__(self): + def __init__(self, logger: logging.Logger | Logger | None = None): self.success_messages: list[BatchEventTypes] = [] self.fail_messages: list[BatchEventTypes] = [] self.exceptions: list[ExceptionInfo] = [] + self.logger = logger @abstractmethod def _prepare(self): @@ -237,6 +239,13 @@ def failure_handler(self, record, exception: ExceptionInfo) -> FailureResponse: exception_string = f"{exception[0]}:{exception[1]}" 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( + "Record processing exception; skipping this record", + exc_info=exception, + ) + self.exceptions.append(exception) self.fail_messages.append(record) return entry @@ -250,6 +259,7 @@ def __init__( event_type: EventType, model: BatchTypeModels | None = None, raise_on_entire_batch_failure: bool = True, + logger: logging.Logger | Logger | None = None, ): """Process batch and partially report failed items @@ -262,6 +272,8 @@ def __init__( raise_on_entire_batch_failure: bool Raise an exception when the entire batch has failed processing. When set to False, partial failures are reported in the response + logger: logging.Logger | Logger | None + Optional Logger instance to output warnings with tracebacks for failed records. Exceptions ---------- @@ -285,7 +297,7 @@ def __init__( EventType.Kafka: KafkaEventRecord, } - super().__init__() + super().__init__(logger=logger) def response(self) -> PartialItemFailureResponse: """Batch items that failed processing, if any""" diff --git a/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py b/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py index 2e680e2f04e..441efc1d288 100644 --- a/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py +++ b/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py @@ -10,6 +10,7 @@ ) if TYPE_CHECKING: + from aws_lambda_powertools.logging import Logger from aws_lambda_powertools.utilities.batch.types import BatchSqsTypeModel logger = logging.getLogger(__name__) @@ -66,7 +67,12 @@ def lambda_handler(event, context: LambdaContext): None, ) - def __init__(self, model: BatchSqsTypeModel | None = None, skip_group_on_error: bool = False): + def __init__( + self, + model: BatchSqsTypeModel | None = None, + skip_group_on_error: bool = False, + logger: logging.Logger | Logger | None = None, + ): """ Initialize the SqsFifoProcessor. @@ -77,12 +83,14 @@ def __init__(self, model: BatchSqsTypeModel | None = None, skip_group_on_error: skip_group_on_error: bool Determines whether to exclusively skip messages from the MessageGroupID that encountered processing failures Default is False. + logger: logging.Logger | Logger | None + Optional Logger instance to output warnings with tracebacks for failed records. """ self._skip_group_on_error: bool = skip_group_on_error self._current_group_id = None self._failed_group_ids: set[str] = set() - super().__init__(EventType.SQS, model) + super().__init__(EventType.SQS, model, logger=logger) def _process_record(self, record): self._current_group_id = record.get("attributes", {}).get("MessageGroupId") diff --git a/tests/functional/batch/required_dependencies/test_utilities_batch.py b/tests/functional/batch/required_dependencies/test_utilities_batch.py index 43c2aa16191..91672d46620 100644 --- a/tests/functional/batch/required_dependencies/test_utilities_batch.py +++ b/tests/functional/batch/required_dependencies/test_utilities_batch.py @@ -861,3 +861,85 @@ async def simple_async_handler(record: SQSRecord): # THEN record is processed successfully using asyncio.run() assert result == {"batchItemFailures": []} assert result == {"batchItemFailures": []} + + +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") + success_record = sqs_event_factory("success") + + def handler(record): + if "fail" in record["body"]: + raise ValueError("intentional failure") + return record["body"] + + test_logger = logging.getLogger("test_logger") + processor = BatchProcessor(event_type=EventType.SQS, logger=test_logger) + + with caplog.at_level(logging.WARNING, logger="test_logger"): + process_partial_response( + event={"Records": [fail_record, success_record]}, + record_handler=handler, + processor=processor, + ) + + warning_records = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warning_records) == 1, f"Expected 1 WARNING log, got {len(warning_records)}" + assert "intentional failure" in warning_records[0].getMessage() or warning_records[0].exc_info is not None + assert warning_records[0].exc_info is not None, "Expected exc_info (traceback) in log record" + assert warning_records[0].exc_info[0] is ValueError + + +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") + + def handler(record): + raise ValueError("intentional failure") + + processor = BatchProcessor(event_type=EventType.SQS, raise_on_entire_batch_failure=False, logger=None) + + with caplog.at_level(logging.WARNING, logger="aws_lambda_powertools.utilities.batch.base"): + process_partial_response( + event={"Records": [fail_record]}, + record_handler=handler, + processor=processor, + ) + + warning_records = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warning_records) == 0, "Expected no WARNING logs when logger is None" + + +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") + short_circuited_record = sqs_event_fifo_factory("would-succeed", "group-1") + + def handler(record): + if "fail" in record["body"]: + raise ValueError("first record failure") + return record["body"] + + test_logger = logging.getLogger("test_logger") + processor = SqsFifoPartialProcessor(logger=test_logger) + processor.raise_on_entire_batch_failure = False + + with caplog.at_level(logging.WARNING, logger="test_logger"): + process_partial_response( + event={"Records": [failing_record, short_circuited_record]}, + record_handler=handler, + processor=processor, + ) + + warning_records = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warning_records) == 1, ( + f"Expected exactly 1 WARNING (real exception only), got {len(warning_records)}: " + + str([r.getMessage() for r in warning_records]) + ) + assert warning_records[0].exc_info[0] is ValueError From acd3734fca4bb7f0336bc755c9ff390a88cda3a6 Mon Sep 17 00:00:00 2001 From: Sujit Date: Thu, 18 Jun 2026 18:37:31 +0530 Subject: [PATCH 3/3] fix: satisfy batch logger mypy checks --- aws_lambda_powertools/utilities/batch/base.py | 18 ++++++++++++++---- .../test_utilities_batch.py | 3 +++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/aws_lambda_powertools/utilities/batch/base.py b/aws_lambda_powertools/utilities/batch/base.py index f51b17db6f5..64c91149e05 100644 --- a/aws_lambda_powertools/utilities/batch/base.py +++ b/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) diff --git a/tests/functional/batch/required_dependencies/test_utilities_batch.py b/tests/functional/batch/required_dependencies/test_utilities_batch.py index 91672d46620..8b255ab1bf5 100644 --- a/tests/functional/batch/required_dependencies/test_utilities_batch.py +++ b/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")