⚡ Bolt: optimize DynamoDB Stream hot path

.jules/bolt.md

@@ -21,3 +21,7 @@
 ## 2026-06-11 - [General] Optimized AWS Service Clients with botocore.config.Config
 **Learning:** Configuring Boto3 clients with `tcp_keepalive=True` and `retries={"max_attempts": 3, "mode": "standard"}` in the `botocore.config.Config` significantly improves connection resilience and reduces latency in AWS Lambda. TCP keep-alive ensures that connections in the pool remain active, avoiding the overhead of re-establishing TCP/TLS handshakes, while the 'standard' retry mode provides more robust exponential backoff.
 **Action:** Always use a centralized `botocore.config.Config` when instantiating Boto3 resources or clients in Lambda templates to optimize performance and reliability.
+
+## 2026-07-01 - [Stream] Bypassing Redundant Validation in Stream Hot Path
+**Learning:** In high-throughput DynamoDB Stream handlers, validating against multiple Pydantic models (e.g., `SourceItem` then `DestinationItem`) adds significant CPU overhead. Since AWS Lambda Powertools already deserializes the DynamoDB event into plain Python dictionaries, we can validate directly into the destination model. Furthermore, for `REMOVE` events, direct dictionary access to keys avoids the overhead of model instantiation entirely.
+**Action:** Minimize the number of model instantiations in stream and batch processing hot paths by validating directly against the final required model and using direct dictionary access for simple key lookups.

templates/stream/handler.py

@@ -6,7 +6,7 @@
 from pydantic import ValidationError
 
 from templates.repository import Repository
-from templates.stream.models import DestinationItem, SourceItem
+from templates.stream.models import DestinationItem
 from templates.stream.settings import Settings
 
 settings = Settings()
@@ -30,18 +30,20 @@ def __init__(self, repository: Repository) -> None:
         self._repository = repository
 
     @tracer.capture_method
-    def _process(self, item: SourceItem) -> DestinationItem | None:
-        """Transform a source item into a destination item.
+    def _process(self, item_dict: dict) -> DestinationItem | None:
+        """Transform a source item dictionary into a destination item.
 
         Args:
-            item: The source item to process.
+            item_dict: The source item dictionary to process.
 
         Returns:
             A `DestinationItem` on success, or `None` if validation fails.
         """
+        # Optimize performance by validating directly into the destination model,
+        # bypassing redundant intermediate validation steps.
         try:
             # TODO: process here
-            return DestinationItem.model_validate(item, from_attributes=True)
+            return DestinationItem.model_validate(item_dict)
         except ValidationError as exc:
             logger.error("DestinationItem validation failed", exc_info=exc)
             return None
@@ -63,13 +65,17 @@ def handle_record(self, record: DynamoDBRecord) -> None:
         event_name = record.event_name
 
         if event_name and event_name.name in ("INSERT", "MODIFY"):
-            item = self._process(SourceItem.model_validate(record.dynamodb.new_image))
+            # Powertools already deserializes the DynamoDB NewImage into a plain dict
+            item = self._process(record.dynamodb.new_image)
             if item is None:
                 raise ValueError("Failed to process record into DestinationItem")
             self._repository.put_item(item.model_dump())
         elif event_name and event_name.name == "REMOVE":
-            plain_keys = SourceItem.model_validate(record.dynamodb.keys)
-            self._repository.delete_item(plain_keys.id)
+            # Direct access to the key dictionary avoids redundant model instantiation on the removal hot path
+            item_id = record.dynamodb.keys.get("id")
+            if not item_id:
+                raise ValueError("Missing 'id' in REMOVE record keys")
+            self._repository.delete_item(item_id)
 
 
 handler = Handler(repository)