Decorators#

AWS Lambda SQS message processing decorators.

This module provides decorators for processing SQS messages in Lambda functions with automatic error handling and batch failure reporting using AWS Lambda’s partial batch response feature.

Type Aliases:

MessageHandler: Function that processes individual SQS messages/records.
LambdaHandler: Standard AWS Lambda handler function signature.
BatchResponse: Lambda response format for batch item failures.
DecoratedHandler: Lambda handler that returns batch failure information.

core_aws.services.lambdas.decorators.process_sqs_batch(message_handler: Callable[[EventRecord | Dict[str, Any]], Any], logger: Logger, post_process_fcn: Callable[[], None] | None = None) → Callable[[Callable[[Dict[str, Any], LambdaContext], Any]], Callable[[Dict[str, Any], LambdaContext], Dict[str, List[Dict[str, str]]]]][source]#

Process SQS messages with partial batch failure reporting (RECOMMENDED). This decorator implements AWS Lambda’s partial batch response feature for SQS, which is the recommended and safer approach compared to manual message deletion. Failed messages are automatically returned to the queue for retry, while successful messages are deleted by Lambda.

IMPORTANT: You must enable “Report batch item failures” in your Lambda trigger configuration for this decorator to work properly.

References

Parameters:

message_handler – Function that processes each individual message (MessageHandler). Receives an EventRecord instance or Dict containing the message data. Should raise an exception if processing fails to mark the message for retry.
logger – Logger instance for tracking message processing and errors.
post_process_fcn – Optional cleanup function called after all messages are processed, regardless of success or failure (e.g., closing database connections, releasing resources). Receives no arguments and returns None.

Returns:

Decorator function that transforms a LambdaHandler into a DecoratedHandler that returns batch failure information.

Example

import logging
from core_aws.services.lambdas.decorators import process_sqs_batch
from core_aws.services.lambdas.events import SqsRecord

logger = logging.getLogger(__name__)

def process_message(record):
    """Process individual SQS message."""
    # Access message content
    message_data = json.loads(record.message)

    # Process the data
    result = do_something(message_data)

    # Raise exception on failure to mark message as failed
    if not result:
        raise ValueError("Processing failed")

def cleanup():
    """Optional cleanup after batch processing."""
    db_connection.close()

@process_sqs_batch(
    message_handler=process_message,
    logger=logger,
    post_process_fcn=cleanup
)
def lambda_handler(event, context):
    """
    Lambda handler - executed before processing messages.
    Body is optional - use for initialization if needed.
    """
    logger.info("Lambda invoked")
    # Optional: initialization code here

# Lambda returns:
# {
#     "batchItemFailures": [
#         {"itemIdentifier": "failed-message-id-1"},
#         {"itemIdentifier": "failed-message-id-2"}
#     ]
# }

Benefits:

No manual deletion: Lambda handles message deletion automatically
Safer: Failed messages stay in queue for retry
Simpler: No need to manage SQS client or queue URLs
AWS recommended: Official pattern for SQS + Lambda
Better visibility: CloudWatch metrics show batch failures

Configuration:

Enable “Report batch item failures” in Lambda trigger: 1. Go to Lambda console > Your function > Configuration > Triggers 2. Edit SQS trigger 3. Expand “Additional settings” 4. Enable “Report batch item failures” 5. Save changes

Decorators#

This Page