forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgetting_started_dynamodb_context_manager.py
More file actions
33 lines (26 loc) · 1.11 KB
/
Copy pathgetting_started_dynamodb_context_manager.py
File metadata and controls
33 lines (26 loc) · 1.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import json
from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
DynamoDBRecord,
)
from aws_lambda_powertools.utilities.typing import LambdaContext
processor = BatchProcessor(event_type=EventType.DynamoDBStreams)
tracer = Tracer()
logger = Logger()
@tracer.capture_method
def record_handler(record: DynamoDBRecord):
if record.dynamodb and record.dynamodb.new_image:
logger.info(record.dynamodb.new_image)
message = record.dynamodb.new_image.get("Message")
if message:
payload: dict = json.loads(message)
logger.info(payload)
@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 = processor.process() # kick off processing, return list[tuple]
logger.info(f"Processed ${len(processed_messages)} messages")
return processor.response()