⚡ Bolt: Optimize DynamoDB Stream Handler processing

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,19 @@ 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) -> DestinationItem | None:
+        """Transform a source item dictionary into a destination item.
 
         Args:
-            item: The source item to process.
+            item: The source item dictionary to process.
 
         Returns:
             A `DestinationItem` on success, or `None` if validation fails.
         """
         try:
-            # TODO: process here
-            return DestinationItem.model_validate(item, from_attributes=True)
+            # Optimize: Bypass redundant SourceItem validation by validating the record dictionary
+            # directly against DestinationItem, avoiding double model validation overhead.
+            return DestinationItem.model_validate(item)
         except ValidationError as exc:
             logger.error("DestinationItem validation failed", exc_info=exc)
             return None
@@ -58,18 +59,31 @@ def handle_record(self, record: DynamoDBRecord) -> None:
             record: The DynamoDB Stream record to process.
 
         Raises:
-            ValueError: If the record cannot be transformed into a `DestinationItem`.
+            ValueError: If the record cannot be transformed into a `DestinationItem` or is invalid.
         """
         event_name = record.event_name
 
+        # Ensure record.dynamodb is not None before accessing new_image or keys to prevent runtime errors
+        if not record.dynamodb:
+            logger.error("Record dynamodb property is None", extra={"record": record})
+            raise ValueError("Invalid record: missing dynamodb data")
+
         if event_name and event_name.name in ("INSERT", "MODIFY"):
-            item = self._process(SourceItem.model_validate(record.dynamodb.new_image))
+            if record.dynamodb.new_image is None:
+                raise ValueError("Failed to process record: new_image is None")
+            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())
+            # Optimize: Use .dump() helper method for consistent camelCase and null field exclusion
+            self._repository.put_item(item.dump())
         elif event_name and event_name.name == "REMOVE":
-            plain_keys = SourceItem.model_validate(record.dynamodb.keys)
-            self._repository.delete_item(plain_keys.id)
+            if record.dynamodb.keys is None:
+                raise ValueError("Failed to process REMOVE record: keys is None")
+            item_id = record.dynamodb.keys.get("id")
+            if not item_id:
+                raise ValueError("Failed to process REMOVE record: 'id' is missing in keys")
+            # Optimize: Bypass SourceItem validation for REMOVE events by passing the key string directly
+            self._repository.delete_item(str(item_id))
 
 
 handler = Handler(repository)