Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -14,41 +14,41 @@ class EventBridgeEvent(DictWrapper):
"""

@property
def get_id(self) -> str:
def get_id(self) -> str | None:
"""A unique value is generated for every event. This can be helpful in tracing events as
they move through rules to targets, and are processed."""
# Note: this name conflicts with existing python builtins
return self["id"]
return self.get("id")

@property
def version(self) -> str:
def version(self) -> str | None:
"""By default, this is set to 0 (zero) in all events."""
return self["version"]
return self.get("version")

@property
def account(self) -> str:
def account(self) -> str | None:
"""The 12-digit number identifying an AWS account."""
return self["account"]
return self.get("account")

@property
def time(self) -> str:
def time(self) -> str | None:
"""The event timestamp, which can be specified by the service originating the event.

If the event spans a time interval, the service might choose to report the start time, so
this value can be noticeably before the time the event is actually received.
"""
return self["time"]
return self.get("time")

@property
def region(self) -> str:
def region(self) -> str | None:
"""Identifies the AWS region where the event originated."""
return self["region"]
return self.get("region")

@property
def resources(self) -> list[str]:
def resources(self) -> list[str] | None:
"""This JSON array contains ARNs that identify resources that are involved in the event.
Inclusion of these ARNs is at the discretion of the service."""
return self["resources"]
return self.get("resources")

@property
def source(self) -> str:
Expand All @@ -68,4 +68,4 @@ def detail(self) -> dict[str, Any]:
@property
def replay_name(self) -> str | None:
"""Identifies whether the event is being replayed and what is the name of the replay."""
return self["replay-name"]
return self.get("replay-name")
7 changes: 7 additions & 0 deletions tests/events/eventBridgeEventMinInfo.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
{
"detail-type": "My custom event",
"source": "my-custom-source",
"detail": {
"custom_field": "cool"
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,3 +18,24 @@ def test_event_bridge_event():
assert parsed_event.detail_type == raw_event["detail-type"]
assert parsed_event.detail == raw_event["detail"]
assert parsed_event.replay_name == "replay_archive"


def test_event_bridge_event_with_minimum_info():
"""Testing a custom event with minimum required information.

ref: https://docs.aws.amazon.com/eventbridge/latest/ref/events-structure.html#eb-custom-event
"""
raw_event = load_event("eventBridgeEventMinInfo.json")
parsed_event = EventBridgeEvent(raw_event)

assert parsed_event.source == raw_event["source"]
assert parsed_event.detail_type == raw_event["detail-type"]
assert parsed_event.detail == raw_event["detail"]

assert parsed_event.get_id is None
assert parsed_event.version is None
assert parsed_event.account is None
assert parsed_event.time is None
assert parsed_event.region is None
assert parsed_event.resources is None
assert parsed_event.replay_name is None