Bug: SqsFifoPartialProcessor fails the whole batch when a short-circuited record doesn't match the model #8661

if fail_entire_batch or fail_group_id:
    return self.failure_handler(
        record=self._to_batch_type(record, event_type=self.event_type, model=self.model),
        exception=...,
    )

from pydantic import BaseModel, Json

from aws_lambda_powertools.utilities.batch import SqsFifoPartialProcessor, process_partial_response
from aws_lambda_powertools.utilities.parser.models import SqsRecordModel


class OrderBody(BaseModel):
    order_id: int


class Order(SqsRecordModel):
    body: Json[OrderBody]


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


def record_handler(record: Order):
    if record.body.order_id == 1:
        raise RuntimeError("failed to process order")


event = {
    "Records": [
        make_record("1", '{"order_id": 1}', "group-1"),  # fails in the handler
        make_record("2", '{"unexpected": 1}', "group-1"),  # doesn't match the model
        make_record("3", '{"order_id": 3}', "group-2"),
    ],
}

for skip_group_on_error in (False, True):
    processor = SqsFifoPartialProcessor(model=Order, skip_group_on_error=skip_group_on_error)
    # Expected: batchItemFailures for 1 and 2 (and 3 when skip_group_on_error=False)
    # Actual: pydantic ValidationError for record 2 escapes and fails the invocation
    print(process_partial_response(event=event, record_handler=record_handler, processor=processor))