Merge pull request #211 from heitorlessa/feat/eventbridge-replay-support

feat: include new EventBridge replay-name field in parser and data_classes utilities

Author: heitorlessa

aws_lambda_powertools/utilities/data_classes/event_bridge_event.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
@@ -62,3 +62,8 @@ def detail_type(self) -> str:
     def detail(self) -> Dict[str, Any]:
         """A JSON object, whose content is at the discretion of the service originating the event. """
         return self["detail"]
+
+    @property
+    def replay_name(self) -> Optional[str]:
+        """Identifies whether the event is being replayed and what is the name of the replay."""
+        return self["replay-name"]

aws_lambda_powertools/utilities/parser/models/event_bridge.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel, Field
 
@@ -14,3 +14,4 @@ class EventBridgeModel(BaseModel):
     resources: List[str]
     detail_type: str = Field(None, alias="detail-type")
     detail: Dict[str, Any]
+    replay_name: Optional[str] = Field(None, alias="replay-name")

tests/events/eventBridgeEvent.json

@@ -12,5 +12,6 @@
   "detail": {
     "instance_id": "i-1234567890abcdef0",
     "state": "terminated"
-  }
-}
+  },
+  "replay-name": "replay_archive"
+}

tests/functional/parser/test_eventbridge.py

@@ -27,6 +27,7 @@ def handle_eventbridge_no_envelope(event: MyAdvancedEventbridgeBusiness, _: Lamb
     assert event.resources == ["arn:aws:ec2:us-west-1:123456789012:instance/i-1234567890abcdef0"]
     assert event.source == "aws.ec2"
     assert event.detail_type == "EC2 Instance State-change Notification"
+    assert event.replay_name == "replay_archive"
 
 
 def test_handle_eventbridge_trigger_event():

tests/functional/test_lambda_trigger_events.py

@@ -370,6 +370,7 @@ def test_event_bridge_event():
     assert event.source == event["source"]
     assert event.detail_type == event["detail-type"]
     assert event.detail == event["detail"]
+    assert event.replay_name == "replay_archive"
 
 
 def test_s3_trigger_event():