docs(batch_processing): snippets split, improved, and lint

examples/batch_processing/sam/dynamodb_batch_processing.yaml

@@ -0,0 +1,66 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: partial batch response sample
+
+Globals:
+  Function:
+    Timeout: 5
+    MemorySize: 256
+    Runtime: python3.10
+    Tracing: Active
+    Environment:
+      Variables:
+        LOG_LEVEL: INFO
+        POWERTOOLS_SERVICE_NAME: hello
+
+Resources:
+  HelloWorldFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: app.lambda_handler
+      CodeUri: hello_world
+      Policies:
+        # Lambda Destinations require additional permissions
+        # to send failure records from Kinesis/DynamoDB
+        - Version: "2012-10-17"
+          Statement:
+            Effect: "Allow"
+            Action:
+              - sqs:GetQueueAttributes
+              - sqs:GetQueueUrl
+              - sqs:SendMessage
+            Resource: !GetAtt SampleDLQ.Arn
+      Events:
+        DynamoDBStream:
+          Type: DynamoDB
+          Properties:
+            Stream: !GetAtt SampleTable.StreamArn
+            StartingPosition: LATEST
+            MaximumRetryAttempts: 2
+            DestinationConfig:
+              OnFailure:
+                Destination: !GetAtt SampleDLQ.Arn
+            FunctionResponseTypes:
+              - ReportBatchItemFailures
+
+  SampleDLQ:
+    Type: AWS::SQS::Queue
+
+  SampleTable:
+    Type: AWS::DynamoDB::Table
+    Properties:
+      BillingMode: PAY_PER_REQUEST
+      AttributeDefinitions:
+        - AttributeName: pk
+          AttributeType: S
+        - AttributeName: sk
+          AttributeType: S
+      KeySchema:
+        - AttributeName: pk
+          KeyType: HASH
+        - AttributeName: sk
+          KeyType: RANGE
+      SSESpecification:
+        SSEEnabled: true
+      StreamSpecification:
+        StreamViewType: NEW_AND_OLD_IMAGES

examples/batch_processing/sam/kinesis_batch_processing.yaml

@@ -0,0 +1,53 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: partial batch response sample
+
+Globals:
+  Function:
+    Timeout: 5
+    MemorySize: 256
+    Runtime: python3.10
+    Tracing: Active
+    Environment:
+      Variables:
+        LOG_LEVEL: INFO
+        POWERTOOLS_SERVICE_NAME: hello
+
+Resources:
+  HelloWorldFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: app.lambda_handler
+      CodeUri: hello_world
+      Policies:
+        # Lambda Destinations require additional permissions
+        # to send failure records to DLQ from Kinesis/DynamoDB
+        - Version: "2012-10-17"
+          Statement:
+            Effect: "Allow"
+            Action:
+              - sqs:GetQueueAttributes
+              - sqs:GetQueueUrl
+              - sqs:SendMessage
+            Resource: !GetAtt SampleDLQ.Arn
+      Events:
+        KinesisStream:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt SampleStream.Arn
+            BatchSize: 100
+            StartingPosition: LATEST
+            MaximumRetryAttempts: 2
+            DestinationConfig:
+              OnFailure:
+                Destination: !GetAtt SampleDLQ.Arn
+            FunctionResponseTypes:
+              - ReportBatchItemFailures
+
+  SampleDLQ:
+    Type: AWS::SQS::Queue
+
+  SampleStream:
+    Type: AWS::Kinesis::Stream
+    Properties:
+      ShardCount: 1

examples/batch_processing/sam/sqs_batch_processing.yaml

@@ -0,0 +1,42 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: partial batch response sample
+
+Globals:
+  Function:
+    Timeout: 5
+    MemorySize: 256
+    Runtime: python3.10
+    Tracing: Active
+    Environment:
+      Variables:
+        LOG_LEVEL: INFO
+        POWERTOOLS_SERVICE_NAME: hello
+
+Resources:
+  HelloWorldFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: app.lambda_handler
+      CodeUri: hello_world
+      Policies:
+        - SQSPollerPolicy:
+            QueueName: !GetAtt SampleQueue.QueueName
+      Events:
+        Batch:
+          Type: SQS
+          Properties:
+            Queue: !GetAtt SampleQueue.Arn
+            FunctionResponseTypes:
+              - ReportBatchItemFailures
+
+  SampleDLQ:
+    Type: AWS::SQS::Queue
+
+  SampleQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      VisibilityTimeout: 30 # Fn timeout * 6
+      RedrivePolicy:
+        maxReceiveCount: 2
+        deadLetterTargetArn: !GetAtt SampleDLQ.Arn

examples/batch_processing/src/advanced_accessing_lambda_context.py

@@ -1,4 +1,3 @@
-import json
 from typing import Optional
 
 from aws_lambda_powertools import Logger, Tracer
@@ -17,11 +16,9 @@
 
 @tracer.capture_method
 def record_handler(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
-    payload: str = record.body
-    if payload:
-        item: dict = json.loads(payload)
-        logger.info(item)
-    ...
+    if lambda_context is not None:
+        remaining_time = lambda_context.get_remaining_time_in_millis()
+        logger.info(remaining_time)
 
 
 @logger.inject_lambda_context

examples/batch_processing/src/advanced_accessing_lambda_context_decorator.py

@@ -0,0 +1,28 @@
+from typing import Optional
+
+from aws_lambda_powertools import Logger, Tracer
+from aws_lambda_powertools.utilities.batch import (
+    BatchProcessor,
+    EventType,
+    batch_processor,
+)
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+processor = BatchProcessor(event_type=EventType.SQS)
+tracer = Tracer()
+logger = Logger()
+
+
+@tracer.capture_method
+def record_handler(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
+    if lambda_context is not None:
+        remaining_time = lambda_context.get_remaining_time_in_millis()
+        logger.info(remaining_time)
+
+
+@logger.inject_lambda_context
+@tracer.capture_lambda_handler
+@batch_processor(record_handler=record_handler, processor=processor)
+def lambda_handler(event, context: LambdaContext):
+    return processor.response()

examples/batch_processing/src/advanced_accessing_lambda_context_manager.py

@@ -0,0 +1,27 @@
+from typing import Optional
+
+from aws_lambda_powertools import Logger, Tracer
+from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+processor = BatchProcessor(event_type=EventType.SQS)
+tracer = Tracer()
+logger = Logger()
+
+
+@tracer.capture_method
+def record_handler(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
+    if lambda_context is not None:
+        remaining_time = lambda_context.get_remaining_time_in_millis()
+        logger.info(remaining_time)
+
+
+@logger.inject_lambda_context
+@tracer.capture_lambda_handler
+def lambda_handler(event, context: LambdaContext):
+    batch = event["Records"]
+    with processor(records=batch, handler=record_handler, lambda_context=context):
+        result = processor.process()
+
+    return result

examples/batch_processing/src/context_manager_access.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import json
+from typing import List, Tuple
+
+from typing_extensions import Literal
+
+from aws_lambda_powertools import Logger, Tracer
+from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+processor = BatchProcessor(event_type=EventType.SQS)
+tracer = Tracer()
+logger = Logger()
+
+
+@tracer.capture_method
+def record_handler(record: SQSRecord):
+    payload: str = record.body
+    if payload:
+        item: dict = json.loads(payload)
+        logger.info(item)
+
+
+@logger.inject_lambda_context
+@tracer.capture_lambda_handler
+def lambda_handler(event, context: LambdaContext):
+    batch = event["Records"]
+    with processor(records=batch, handler=record_handler):
+        processed_messages: List[Tuple] = processor.process()
+
+    for message in processed_messages:
+        status: Literal["success"] | Literal["fail"] = message[0]
+        record: SQSRecord = message[2]
+
+        logger.info(status, record=record)
+
+    return processor.response()

examples/batch_processing/src/custom_partial_processor.py

@@ -0,0 +1,76 @@
+import os
+import sys
+from random import randint
+from typing import Any
+
+import boto3
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.batch import (
+    BasePartialBatchProcessor,
+    EventType,
+    process_partial_response,
+)
+
+table_name = os.getenv("TABLE_NAME", "table_not_found")
+
+logger = Logger()
+
+
+class MyPartialProcessor(BasePartialBatchProcessor):
+    """
+    Process a record and stores successful results at a Amazon DynamoDB Table
+
+    Parameters
+    ----------
+    table_name: str
+        DynamoDB table name to write results to
+    """
+
+    def __init__(self, table_name: str):
+        self.table_name = table_name
+
+        super().__init__(event_type=EventType.SQS)
+
+    def _prepare(self):
+        # It's called once, *before* processing
+        # Creates table resource and clean previous results
+        self.ddb_table = boto3.resource("dynamodb").Table(self.table_name)
+        self.success_messages.clear()
+
+    def _clean(self):
+        # It's called once, *after* closing processing all records (closing the context manager)
+        # Here we're sending, at once, all successful messages to a ddb table
+        with self.ddb_table.batch_writer() as batch:
+            for result in self.success_messages:
+                batch.put_item(Item=result)
+
+    def _process_record(self, record):
+        # It handles how your record is processed
+        # Here we're keeping the status of each run
+        # where self.handler is the record_handler function passed as an argument
+        try:
+            result = self.handler(record)  # record_handler passed to decorator/context manager
+            return self.success_handler(record, result)
+        except Exception as exc:
+            logger.error(exc)
+            return self.failure_handler(record, sys.exc_info())
+
+    def success_handler(self, record, result: Any):
+        entry = ("success", result, record)
+        self.success_messages.append(record)
+        return entry
+
+    async def _async_process_record(self, record: dict):
+        raise NotImplementedError()
+
+
+processor = MyPartialProcessor(table_name)
+
+
+def record_handler(record):
+    return randint(0, 100)
+
+
+def lambda_handler(event, context):
+    return process_partial_response(event=event, record_handler=record_handler, processor=processor, context=context)

examples/batch_processing/src/disable_tracing.py

@@ -0,0 +1,28 @@
+import json
+
+from aws_lambda_powertools import Logger, Tracer
+from aws_lambda_powertools.utilities.batch import (
+    BatchProcessor,
+    EventType,
+    process_partial_response,
+)
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+processor = BatchProcessor(event_type=EventType.SQS)
+tracer = Tracer()
+logger = Logger()
+
+
+@tracer.capture_method(capture_response=False)
+def record_handler(record: SQSRecord):
+    payload: str = record.body
+    if payload:
+        item: dict = json.loads(payload)
+        logger.info(item)
+
+
+@logger.inject_lambda_context
+@tracer.capture_lambda_handler
+def lambda_handler(event, context: LambdaContext):
+    return process_partial_response(event=event, record_handler=record_handler, processor=processor, context=context)