Repository: aws-powertools/powertools-lambda-python

Issue:
Feature request: Batch processor should fail early if event payload is incorrect 
### Expected Behaviour

@heitorlessa this is a continuation of [23777](https://github.com/awslabs/aws-lambda-powertools-python/issues/2377)

When using the `BatchProcessor` with `process_partial_response` from [here](https://awslabs.github.io/aws-lambda-powertools-python/2.16.1/utilities/batch/) I have noticed that the event does not have to respect the model and it can return an empty list for the `batchItemFailures` which will tell AWS to delete the messages from the queue in case of SQS triggers. 

I'd expect the behavior to be fail-first fail-fast. If I got a bunch of messages to my Lambda that do not have the format of an `EventType.SQS` that should fail right away, in my opinion. Passing a random `dict` to the `process_partial_response` should not be considered a valid event.



### Current Behaviour

The current implementation of `process_partial_response` accepts any `dict` passed in. Since it used the `dict` method `.get` to get the `Records` which is a key in SQS type events. But returns a default empty list in case the key doesn't exist.  Which in turn will tell the context manager to run the processor on an empty list of records, which will always return success.


![image](https://github.com/awslabs/aws-lambda-powertools-python/assets/131292734/b6c64d1f-384e-4d32-8cb7-cdd83b304e44)


### Code snippet

```python
from aws_lambda_powertools.utilities.batch import (
    BatchProcessor, EventType, process_partial_response
)
from aws_lambda_powertools.utilities.typing import LambdaContext
import pydantic


class Event(pydantic.BaseModel):
    foo: str
    bar: str


def record_handler(record: Event):
    print(record)

    if record.foo == "barr":
        raise ValueError


def handler(event: dict, context: LambdaContext):

    return process_partial_response(
        event=event,
        record_handler=record_handler,
        processor=BatchProcessor(
            event_type=EventType.SQS,
            model=Event,
        ),
        context=context
    )


# This is an invalid event since it's not of the type SQS
# But the error about missing SQS model field is only thrown after the record_handler is called
# It should be at the very beginning, before any business logic occurs
print(handler({'Records': [{"foo": "bar", "bar": "baz"}, {"foo": "barr", "bar": "bazz"}]}, object))

# This is not valid, but does not throw an error
# And the batchItemFailures returns an empty list which is not accurate.
print(handler({"foo": "bar", "bar": "baz"}, object))

# This throws ValueError: Invalid event format. Please ensure batch event is a valid SQS event.
print(handler(["foo", "bar", "bar", "baz"], object))
```


### Possible Solution

I'm sure there are a variety of reasons for why this is the way it is that I'm not considering. One obvious answer is that SQS event types will always have the 'Records` key hence anything that doesn't have it should answer the question: "How did you get here and should you even be considered?"

But I believe that we can be a bit more strict on enforcing that the `dict` passed is indeed of the type SQS, DynamoDB stream or Kinesis before doing any processing.

For example, in the snippet above, I pass in a `dict` with the `Records` key, but the rest of the event does not have the remaining SQS keys. When the `record_handler` fails, the batch processor looks for the `messageId` field, which doesn't exist. I believe if this was done upfront and fail immediately it would be a better developer experience. 

What's everyone's thoughts?

### Steps to Reproduce

Snippet above should be enough to demonstrate all scenarios brought up.

### Powertools for AWS Lambda (Python) version

latest

### AWS Lambda function runtime

3.10

### Packaging format used

Lambda Layers

### Debugging logs

_No response_

Discussion:
@heitorlessa: Thanks a lot for creating and giving more context @alexandreczg - I'll look into it first thing tomorrow morning (Amsterdam time).@heitorlessa: Going through this now (operational task took priority) - I removed the `bug` label for now as I'm leaning towards defensive programming from self-inflicted errors here.

I see two ways we can provide a better experience here - which one makes it more appealing to you?

* **A) Fail fast**. We can raise a ValueError if the event was intentionally modified (`if not records: raise ValueError("Invalid event format)`).
    * **Pros**: Immediate feedback although you know what you're doing upfront either way.
    * **Cons**: Lambda Event Source Mapping (a.k.a managed poller) [will scale down the rate messages are polled and your function is invoked](https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html#services-sqs-backoff-strategy). This can take a while to get back to "normal rate" (your mileage varies).

* **B) Warn**. Warn that somehow we received no Records and proceed like we would (not calling your record handler). 
    * **Pros**: Doesn't interfere with Lambda Event Source Mapping. Your tests and actual prod use will continue to partially fail if you have an actual poison pill (malformed record, bad record input).
    * **Cons**: As you pointed out, batch messages will be deleted from the queue because you intentionally didn't pass any to be processed. This means data loss.

While I'd argue this is self-inflicted, I do see the value in failing fast for a faster feedback loop locally when testing. Given the criticality of how many transactions we process a week, I don't feel comfortable in validating the schema because we can't guarantee these event sources won't change, leading to widespread incidents. IF a record is malformed, it'd be caught later in the process whether you use Pydantic or not.

Knowing `Records` wouldn't be empty in prod, I feel comfortable with checking if `Records` is empty for a faster feedback loop locally, where you might be trying custom middlewares that mutate events and accidentally pass the wrong ones, or intentionally testing the Batch processing feature would call your record handler function from an invalid payload.

Happy to guide you on making this first contribution and jump on a call to get your dev env setup :) 

Let us know!

@alexandreczg: Took me a while to carefully digest both options presented and the merits of each. I think there is a lot of value on your arguments. 

I too agree with not throwing an error, it makes sense to me now. Option B) would be interesting to explore in order for the quick feedback loop on local development, especially for the AWS beginner, kind of nudging them in the right direction. 

I'd love to jump on a call at your convenience!@heitorlessa: Awesome, B it is (my preference also). Send an email to aws-lambda-powertools-feedback at amazon dot com, and any of us will reply with a link that you schedule a slot ;)

If you'd like a head start, we have a pre-configured Cloud IDE you can use, along with steps you could use if you want to do this locally (make sure you've got a virtual env):  https://github.com/awslabs/aws-lambda-powertools-python/blob/develop/CONTRIBUTING.md#contributing-via-pull-requests

Looking forward to it!@leandrodamascena: Hi @alexandreczg! I'm checking if you're still interested in submitting this PR. If not, no problem, I'm scheduling a revisit in 1 month and we'll implement option B.@dreamorosi: We have implemented this in Powertools for AWS Lambda (TypeScript) and ended up going [for option B](https://github.com/aws-powertools/powertools-lambda-typescript/blob/a4846afd09c29032f79b79cfa1410675440d7dae/packages/batch/src/processPartialResponse.ts#L81-L83), raising an exception if the event doesn't have shape:

```json
{
  "Records": []
}
```

For the sake of consistency, we should probably go this route here as well.

> [!Note]
> This issue is open for contributions! If you are interested in picking this up please leave a comment to claim it. A maintainer will assign the issue to you and help you with reviewing your PR. Likewise, if you have any questions before starting, feel free to leave a comment below.@anafalcao: I'll work on it and open a PR soon.@anafalcao: We will raise an exception in this case to ensure consistency with the TypeScript version and match the functionality across both runtimes.@github-actions: ### ⚠️COMMENT VISIBILITY WARNING⚠️
This issue is now closed. Please be mindful that future comments are hard for our team to see.

If you need more assistance, please either tag a [team member](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/MAINTAINERS.md#current-maintainers) or open a new issue that references this one.

If you wish to keep having a conversation with other community members under this issue feel free to do so.@github-actions: This is now released under [3.7.0](https://github.com/aws-powertools/powertools-lambda-python/releases/tag/v3.7.0) version!

Solution patch:
diff --git a/aws_lambda_powertools/utilities/batch/decorators.py b/aws_lambda_powertools/utilities/batch/decorators.py
index 0cba41f98..2b9f5433e 100644
--- a/aws_lambda_powertools/utilities/batch/decorators.py
+++ b/aws_lambda_powertools/utilities/batch/decorators.py
@@ -12,6 +12,7 @@ from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
 )
+from aws_lambda_powertools.utilities.batch.exceptions import UnexpectedBatchTypeError
 from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning
 
 if TYPE_CHECKING:
@@ -204,6 +205,11 @@ def process_partial_response(
     """
     try:
         records: list[dict] = event.get("Records", [])
+        if not records or not isinstance(records, list):
+            raise UnexpectedBatchTypeError(
+                "Unexpected batch event type. Possible values are: SQS, KinesisDataStreams, DynamoDBStreams",
+            )
+
     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 +274,11 @@ def async_process_partial_response(
     """
     try:
         records: list[dict] = event.get("Records", [])
+        if not records or not isinstance(records, list):
+            raise UnexpectedBatchTypeError(
+                "Unexpected batch event type. Possible values are: SQS, KinesisDataStreams, DynamoDBStreams",
+            )
+
     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
diff --git a/aws_lambda_powertools/utilities/batch/exceptions.py b/aws_lambda_powertools/utilities/batch/exceptions.py
index c93b96a8f..87a2df22d 100644
--- a/aws_lambda_powertools/utilities/batch/exceptions.py
+++ b/aws_lambda_powertools/utilities/batch/exceptions.py
@@ -38,6 +38,12 @@ class BatchProcessingError(BaseBatchProcessingError):
         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"""
+
+    pass
+
+
 class SQSFifoCircuitBreakerError(Exception):
     """
     Signals a record not processed due to the SQS FIFO processing being interrupted


Test patch:
diff --git a/tests/functional/batch/required_dependencies/test_utilities_batch.py b/tests/functional/batch/required_dependencies/test_utilities_batch.py
index 9327a7d70..4c91dd54a 100644
--- a/tests/functional/batch/required_dependencies/test_utilities_batch.py
+++ b/tests/functional/batch/required_dependencies/test_utilities_batch.py
@@ -15,7 +15,7 @@ from aws_lambda_powertools.utilities.batch import (
     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,56 @@ 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 event 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 event type. Possible values are: SQS, KinesisDataStreams, DynamoDBStreams" in str(
+        exc_info.value,
+    )