Bug: batch processor crashes with AttributeError when a model validator raises a non-ValueError #8660

Expected Behaviour

When converting a record to the customer's model fails, the record should be reported in batchItemFailures and the rest of the batch should be reported normally, the same as the poison-pill handling added in #2091.

Current Behaviour

#2091 only covers Pydantic ValidationErrors. Pydantic v2 only wraps ValueError and AssertionError raised by validators; any other exception (e.g. a KeyError/TypeError from a field_validator that parses the body) propagates as-is. In that case BatchProcessor._process_record falls through to:

return self.failure_handler(record=data, exception=sys.exc_info())

with data still None, because _to_batch_type never returned. None is appended to fail_messages, and _clean() then crashes while collecting failures:

  File ".../aws_lambda_powertools/utilities/batch/base.py", line 365, in _collect_sqs_failures
    msg_id = msg.message_id
             ^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'message_id'

So a single malformed record fails the whole invocation, the original exception is hidden behind the AttributeError, and every record in the batch (including the ones that succeeded) is retried. The same applies to Kinesis, DynamoDB Streams and AsyncBatchProcessor.

Code snippet

import json

from pydantic import field_validator

from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, process_partial_response
from aws_lambda_powertools.utilities.parser.models import SqsRecordModel


class Order(SqsRecordModel):
    @field_validator("body")
    @classmethod
    def parse_order_id(cls, value):
        return json.loads(value)["order_id"]  # KeyError when order_id is missing


def make_record(message_id, body):
    return {
        "messageId": message_id,
        "receiptHandle": "handle",
        "body": body,
        "attributes": {
            "ApproximateReceiveCount": "1",
            "SentTimestamp": "1",
            "SenderId": "sender",
            "ApproximateFirstReceiveTimestamp": "1",
        },
        "messageAttributes": {},
        "md5OfBody": "md5",
        "eventSource": "aws:sqs",
        "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:queue",
        "awsRegion": "us-east-1",
    }


event = {"Records": [make_record("1", '{"order_id": 1}'), make_record("2", '{"unexpected": 1}')]}
processor = BatchProcessor(event_type=EventType.SQS, model=Order)

# Expected: {'batchItemFailures': [{'itemIdentifier': '2'}]}
# Actual: AttributeError: 'NoneType' object has no attribute 'message_id'
print(process_partial_response(event=event, record_handler=lambda record: record.body, processor=processor))

Possible Solution

If data is None when the exception is caught, the failure came from converting the record, so it can go through the existing _register_model_validation_error_record(record) path, which converts the raw record to the event source data class and registers it with the original sys.exc_info():

if model in (self.model, model_name) or data is None:
    return self._register_model_validation_error_record(record)

in both BatchProcessor._process_record and AsyncBatchProcessor._async_process_record. I can open a PR with this and regression tests for the sync and async processors if that works for you.

Steps to Reproduce

Run the snippet above.

Powertools for AWS Lambda (Python) version

latest (develop @ 45f7a4d)

AWS Lambda function runtime

3.11

Packaging format used

PyPi

Debugging logs

No response