feat(batch): raise exception for invalid batch event

aws_lambda_powertools/utilities/batch/decorators.py

@@ -12,6 +12,7 @@
     BatchProcessor,
     EventType,
 )
+from aws_lambda_powertools.utilities.batch.exceptions import UnexpectedBatchTypeError
 from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning
 
 if TYPE_CHECKING:
@@ -204,6 +205,9 @@ def handler(event, context):
     """
     try:
         records: list[dict] = event.get("Records", [])
+        if not records or not isinstance(records, list):
+            raise UnexpectedBatchTypeError
+
     except AttributeError:
         event_types = ", ".join(list(EventType.__members__))
         docs = "https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/#processing-messages-from-sqs"  # noqa: E501 # long-line
@@ -268,6 +272,9 @@ def handler(event, context):
     """
     try:
         records: list[dict] = event.get("Records", [])
+        if not records or not isinstance(records, list):
+            raise UnexpectedBatchTypeError
+
     except AttributeError:
         event_types = ", ".join(list(EventType.__members__))
         docs = "https://docs.powertools.aws.dev/lambda/python/latest/utilities/batch/#processing-messages-from-sqs"  # noqa: E501 # long-line

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -38,6 +38,15 @@ def __str__(self):
         return self.format_exceptions(parent_exception_str)
 
 
+class UnexpectedBatchTypeError(BatchProcessingError):
+    """Error thrown by the Batch Processing utility when a partial processor receives an unexpected batch type"""
+
+    def __init__(self):
+        msg = "Unexpected batch type. Possible values are: SQS, KinesisDataStreams, DynamoDBStreams"
+        super().__init__(msg)
+        self.name = "UnexpectedBatchTypeError"
+
+
 class SQSFifoCircuitBreakerError(Exception):
     """
     Signals a record not processed due to the SQS FIFO processing being interrupted

tests/functional/batch/required_dependencies/test_utilities_batch.py

@@ -15,7 +15,7 @@
     batch_processor,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.batch.exceptions import BatchProcessingError
+from aws_lambda_powertools.utilities.batch.exceptions import BatchProcessingError, UnexpectedBatchTypeError
 from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
     DynamoDBRecord,
 )
@@ -708,3 +708,69 @@ def test_async_process_partial_response_invalid_input(async_record_handler: Call
     # WHEN/THEN
     with pytest.raises(ValueError):
         async_process_partial_response(batch, record_handler, processor)
+
+
+@pytest.mark.parametrize(
+    "event",
+    [
+        {},
+        {"Records": None},
+        {"Records": "not a list"},
+    ],
+)
+def test_process_partial_response_raises_unexpected_batch_type(event, record_handler):
+    # GIVEN a batch processor configured for SQS events
+    processor = BatchProcessor(event_type=EventType.SQS)
+
+    # WHEN processing an event with invalid Records
+    with pytest.raises(UnexpectedBatchTypeError) as exc_info:
+        process_partial_response(
+            event=event,
+            record_handler=record_handler,
+            processor=processor,
+        )
+
+    # THEN the correct error message is raised
+    assert "Unexpected batch type. Possible values are: SQS, KinesisDataStreams, DynamoDBStreams" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "event",
+    [
+        {},
+        {"Records": None},
+        {"Records": "not a list"},
+    ],
+)
+async def test_async_process_partial_response_raises_unexpected_batch_type(event, async_record_handler):
+    # GIVEN a batch processor configured for SQS events
+    processor = BatchProcessor(event_type=EventType.SQS)
+
+    # WHEN processing an event with invalid Records asynchronously
+    with pytest.raises(UnexpectedBatchTypeError) as exc_info:
+        await async_process_partial_response(
+            event=event,
+            record_handler=async_record_handler,
+            processor=processor,
+        )
+
+    # THEN the correct error message is raised
+    assert "Unexpected batch type. Possible values are: SQS, KinesisDataStreams, DynamoDBStreams" in str(exc_info.value)
+
+
+def test_process_partial_response_should_not_raise_unexpected_batch_type(record_handler):
+    # GIVEN a valid SQS event structure
+    event = {"Records": [{"messageId": "1", "body": "test"}]}
+    processor = BatchProcessor(event_type=EventType.SQS)
+
+    # WHEN processing the event
+    try:
+        process_partial_response(
+            event=event,
+            record_handler=record_handler,
+            processor=processor,
+        )
+    except UnexpectedBatchTypeError:
+        # THEN no UnexpectedBatchTypeError should be raised
+        pytest.fail("UnexpectedBatchTypeError was raised with a valid event structure!")