chore: remove middlewares module, moving decorator functionality to base and sqs

Author: Tom McCarthy

aws_lambda_powertools/utilities/batch/__init__.py

@@ -4,8 +4,7 @@
 Batch processing utility
 """
 
-from .base import BasePartialProcessor
-from .middlewares import batch_processor, sqs_batch_processor
-from .sqs import PartialSQSProcessor
+from .base import BasePartialProcessor, batch_processor
+from .sqs import PartialSQSProcessor, sqs_batch_processor
 
 __all__ = ("BasePartialProcessor", "PartialSQSProcessor", "batch_processor", "sqs_batch_processor")

aws_lambda_powertools/utilities/batch/base.py

@@ -5,7 +5,9 @@
 """
 
 from abc import ABC, abstractmethod
-from typing import Any, Callable, Iterable, List, Tuple
+from typing import Any, Callable, Dict, Iterable, List, Tuple
+
+from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
 
 
 class BasePartialProcessor(ABC):
@@ -93,3 +95,48 @@ def failure_handler(self, record: Any, exception: Exception):
         self.exceptions.append(exception)
         self.fail_messages.append(record)
         return entry
+
+
+@lambda_handler_decorator
+def batch_processor(
+    handler: Callable, event: Dict, context: Dict, record_handler: Callable, processor: BasePartialProcessor = None
+):
+    """
+    Middleware to handle batch event processing
+
+    Parameters
+    ----------
+    handler: Callable
+        Lambda's handler
+    event: Dict
+        Lambda's Event
+    context: Dict
+        Lambda's Context
+    record_handler: Callable
+        Callable to process each record from the batch
+    processor: PartialSQSProcessor
+        Batch Processor to handle partial failure cases
+
+    Examples
+    --------
+    **Processes Lambda's event with PartialSQSProcessor**
+        >>> from aws_lambda_powertools.utilities.batch import batch_processor
+        >>>
+        >>> def record_handler(record):
+        >>>     return record["body"]
+        >>>
+        >>> @batch_processor(record_handler=record_handler, processor=PartialSQSProcessor())
+        >>> def handler(event, context):
+        >>>     return {"StatusCode": 200}
+
+    Limitations
+    -----------
+    * Async batch processors
+
+    """
+    records = event["Records"]
+
+    with processor(records, record_handler):
+        processor.process()
+
+    return handler(event, context)

aws_lambda_powertools/utilities/batch/middlewares.py

@@ -3,11 +3,12 @@
 """
 Batch SQS utilities
 """
-from typing import List, Optional, Tuple
+from typing import Callable, Dict, List, Optional, Tuple
 
 import boto3
 from botocore.config import Config
 
+from ...middleware_factory import lambda_handler_decorator
 from .base import BasePartialProcessor
 from .exceptions import SQSBatchProcessingError
 
@@ -116,3 +117,58 @@ def _clean(self):
             raise SQSBatchProcessingError(list(self.exceptions))
 
         return delete_message_response
+
+
+@lambda_handler_decorator
+def sqs_batch_processor(
+    handler: Callable,
+    event: Dict,
+    context: Dict,
+    record_handler: Callable,
+    config: Optional[Config] = None,
+    suppress_exception: bool = False,
+):
+    """
+    Middleware to handle SQS batch event processing
+
+    Parameters
+    ----------
+    handler: Callable
+        Lambda's handler
+    event: Dict
+        Lambda's Event
+    context: Dict
+        Lambda's Context
+    record_handler: Callable
+        Callable to process each record from the batch
+    config: Config
+            botocore config object
+    suppress_exception: bool, optional
+        Supress exception raised if any messages fail processing, by default False
+
+    Examples
+    --------
+    **Processes Lambda's event with PartialSQSProcessor**
+        >>> from aws_lambda_powertools.utilities.batch import sqs_batch_processor
+        >>>
+        >>> def record_handler(record):
+        >>>     return record["body"]
+        >>>
+        >>> @sqs_batch_processor(record_handler=record_handler)
+        >>> def handler(event, context):
+        >>>     return {"StatusCode": 200}
+
+    Limitations
+    -----------
+    * Async batch processors
+
+    """
+    config = config or Config()
+    processor = PartialSQSProcessor(config=config, suppress_exception=suppress_exception)
+
+    records = event["Records"]
+
+    with processor(records, record_handler):
+        processor.process()
+
+    return handler(event, context)

aws_lambda_powertools/utilities/batch/sqs.py

@@ -148,7 +148,7 @@ def lambda_handler(event, context):
         stubber.assert_no_pending_responses()
 
 
-@patch("aws_lambda_powertools.utilities.batch.middlewares.PartialSQSProcessor")
+@patch("aws_lambda_powertools.utilities.batch.sqs.PartialSQSProcessor")
 def test_sqs_batch_processor_middleware(
     patched_sqs_processor, sqs_event_factory, record_handler, stubbed_partial_processor
 ):
@@ -251,7 +251,7 @@ def test_partial_sqs_processor_suppressed_exceptions(sqs_event_factory, record_h
     assert partial_processor_suppressed.success_messages == [first_record]
 
 
-@patch("aws_lambda_powertools.utilities.batch.middlewares.PartialSQSProcessor")
+@patch("aws_lambda_powertools.utilities.batch.sqs.PartialSQSProcessor")
 def test_sqs_batch_processor_middleware_suppressed_exception(
     patched_sqs_processor, sqs_event_factory, record_handler, stubbed_partial_processor_suppressed
 ):