feat(data_classes,parser): add EventBridge Custom Event Bus support

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -28,6 +28,7 @@
 from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
 from .event_bridge_event import EventBridgeEvent
+from .event_bridge_v2_event import EventBridgeV2Event
 from .event_source import event_source
 from .kafka_event import KafkaEvent
 from .kinesis_firehose_event import (
@@ -74,6 +75,7 @@
     "ConnectContactFlowEvent",
     "DynamoDBStreamEvent",
     "EventBridgeEvent",
+    "EventBridgeV2Event",
     "KafkaEvent",
     "KinesisFirehoseEvent",
     "KinesisStreamEvent",

aws_lambda_powertools/utilities/data_classes/event_bridge_v2_event.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import base64
+from functools import cached_property
+from typing import Any, Literal
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+OCTET_STREAM = "application/octet-stream"
+
+
+def _is_with_metadata(item: Any) -> bool:
+    return isinstance(item, dict) and "Data" in item and "SystemMetadata" in item
+
+
+class EventBridgeV2SystemMetadata(DictWrapper):
+    """Metadata added by the bus (`SystemMetadata`), only present with the WITH_METADATA transformer"""
+
+    @property
+    def event_id(self) -> str | None:
+        """Unique event id assigned by the bus (`aws:EventId`). It is not the id of the classic envelope."""
+        return self.get("aws:EventId")
+
+    @property
+    def sequence_number(self) -> str | None:
+        """Sequence number of the event within the bus (`aws:SequenceNumber`)."""
+        return self.get("aws:SequenceNumber")
+
+    @property
+    def delivery_type(self) -> Literal["LIVE", "REPLAY"] | None:
+        """Whether the event is delivered as it arrives (`LIVE`) or as part of a replay (`REPLAY`)."""
+        return self.get("aws:DeliveryType")
+
+    @property
+    def ingestion_time(self) -> str | None:
+        """Time the bus ingested the event (`aws:IngestionTime`)."""
+        return self.get("aws:IngestionTime")
+
+    @property
+    def event_group_id(self) -> str | None:
+        """FIFO group id the event belongs to (`EventGroupId`)."""
+        return self.get("EventGroupId")
+
+    @property
+    def deduplication_id(self) -> str | None:
+        """Deduplication id of the event (`DeduplicationId`)."""
+        return self.get("DeduplicationId")
+
+    @property
+    def content_type(self) -> str | None:
+        """Content type of the payload, e.g. `application/json` or `application/eventbridge+json` (`ContentType`)."""
+        return self.get("ContentType")
+
+    @property
+    def source(self) -> str | None:
+        """Event source (`aws:Source`), only present for events published with PutEvents."""
+        return self.get("aws:Source")
+
+    @property
+    def detail_type(self) -> str | None:
+        """Event detail type (`aws:DetailType`), only present for events published with PutEvents."""
+        return self.get("aws:DetailType")
+
+
+class EventBridgeV2Record(DictWrapper):
+    """A single event delivered by an Amazon EventBridge Custom Event Bus"""
+
+    @property
+    def data(self) -> Any:
+        """Event payload: a dict for JSON, bytes for `application/octet-stream`."""
+        data = self["Data"]
+        if self.system_metadata.content_type == OCTET_STREAM and isinstance(data, str):
+            return base64.b64decode(data)
+        return data
+
+    @property
+    def detail(self) -> dict[str, Any] | None:
+        """Shortcut to `data["detail"]` when the event was published with PutEvents, otherwise None."""
+        data = self["Data"]
+        if isinstance(data, dict) and "detail-type" in data and "detail" in data:
+            return data["detail"]
+        return None
+
+    @property
+    def metadata(self) -> dict[str, Any]:
+        """Producer metadata keys sent with PutRawEvents."""
+        return self.get("Metadata") or {}
+
+    @cached_property
+    def system_metadata(self) -> EventBridgeV2SystemMetadata:
+        """Metadata added by the bus; empty when the subscriber doesn't use the WITH_METADATA transformer."""
+        return EventBridgeV2SystemMetadata(self.get("SystemMetadata") or {})
+
+
+class EventBridgeV2Event(DictWrapper):
+    """Amazon EventBridge Custom Event Bus event
+
+    Accepts what a subscriber can deliver to Lambda: a single object or an array, using the
+    WITH_METADATA transformer (`Data`, `Metadata`, `SystemMetadata`) or RAW (classic envelope,
+    PutEvents array or the bare PutRawEvents payload). RAW deliveries carry no marker, so
+    they are exposed with empty metadata.
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-custom-bus.html
+    """
+
+    @cached_property
+    def records(self) -> list[EventBridgeV2Record]:
+        """Events in the delivery. Always a list, even when the bus sent a single object."""
+        items = self._data if isinstance(self._data, list) else [self._data]
+        return [
+            EventBridgeV2Record(
+                item if _is_with_metadata(item) else {"Data": item, "Metadata": {}, "SystemMetadata": {}},
+            )
+            for item in items
+        ]

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

@@ -6,6 +6,7 @@
 from .cloudwatch import CloudWatchLogsEnvelope
 from .dynamodb import DynamoDBStreamEnvelope
 from .event_bridge import EventBridgeEnvelope
+from .event_bridge_v2 import EventBridgeV2Envelope
 from .kafka import KafkaEnvelope
 from .kinesis import KinesisDataStreamEnvelope
 from .kinesis_firehose import KinesisFirehoseEnvelope
@@ -24,6 +25,7 @@
     "CloudWatchLogsEnvelope",
     "DynamoDBStreamEnvelope",
     "EventBridgeEnvelope",
+    "EventBridgeV2Envelope",
     "KinesisDataStreamEnvelope",
     "KinesisFirehoseEnvelope",
     "LambdaFunctionUrlEnvelope",

aws_lambda_powertools/utilities/parser/envelopes/event_bridge_v2.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import EventBridgeV2Model
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import T
+
+logger = logging.getLogger(__name__)
+
+
+def _is_with_metadata(item: Any) -> bool:
+    return isinstance(item, dict) and "Data" in item and "SystemMetadata" in item
+
+
+def _unwrap_put_events(data: Any) -> Any:
+    if isinstance(data, dict) and "detail-type" in data and "detail" in data:
+        return data["detail"]
+    return data
+
+
+class EventBridgeV2Envelope(BaseEnvelope):
+    """EventBridge Custom Event Bus envelope to extract the payload of each record
+
+    Yields `Data` for events published with PutRawEvents, and `Data.detail` for PutEvents.
+    RAW deliveries (classic envelope, PutEvents array, bare payload) are also accepted.
+    """
+
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]:
+        """Parses records found with model provided
+
+        Parameters
+        ----------
+        data : dict
+            Lambda event to be parsed
+        model : type[T]
+            Data model provided to parse after extracting data using envelope
+
+        Returns
+        -------
+        list[T | None]
+            List of payloads parsed with model provided
+        """
+        items = data if isinstance(data, list) else [data]
+
+        payloads: list[Any] = []
+        for item in items:
+            if _is_with_metadata(item):
+                logger.debug(f"Parsing incoming data with EventBridge V2 model {EventBridgeV2Model}")
+                payload = EventBridgeV2Model.model_validate(item).root[0].data
+            else:
+                payload = item
+            if isinstance(payload, bytes):
+                # application/octet-stream is delivered as bytes; let the JSON path of _parse handle it
+                payload = payload.decode()
+            payloads.append(_unwrap_put_events(payload))
+
+        logger.debug(f"Parsing EventBridge V2 payloads with {model}")
+        return [self._parse(data=payload, model=model) for payload in payloads]

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -77,6 +77,7 @@
         DynamoDBStreamRecordModel,
     )
     from .event_bridge import EventBridgeModel
+    from .event_bridge_v2 import EventBridgeV2Model, EventBridgeV2RecordModel, EventBridgeV2SystemMetadataModel
     from .kafka import (
         KafkaBaseEventModel,
         KafkaMskEventModel,
@@ -199,6 +200,9 @@
     "DynamoDBStreamModel": "dynamodb",
     "DynamoDBStreamRecordModel": "dynamodb",
     "EventBridgeModel": "event_bridge",
+    "EventBridgeV2Model": "event_bridge_v2",
+    "EventBridgeV2RecordModel": "event_bridge_v2",
+    "EventBridgeV2SystemMetadataModel": "event_bridge_v2",
     "KafkaBaseEventModel": "kafka",
     "KafkaMskEventModel": "kafka",
     "KafkaRecordModel": "kafka",
@@ -311,6 +315,9 @@ def __dir__() -> list[str]:
     "AppSyncEventsModel",
     "DynamoDBStreamModel",
     "EventBridgeModel",
+    "EventBridgeV2Model",
+    "EventBridgeV2RecordModel",
+    "EventBridgeV2SystemMetadataModel",
     "DynamoDBStreamChangedRecordModel",
     "DynamoDBStreamLambdaOnFailureDestinationModel",
     "DynamoDBStreamRecordModel",

aws_lambda_powertools/utilities/parser/models/event_bridge_v2.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import base64
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, RootModel, model_validator
+
+OCTET_STREAM = "application/octet-stream"
+
+
+class EventBridgeV2SystemMetadataModel(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    event_id: str | None = Field(None, alias="aws:EventId", description="Unique event id assigned by the bus.")
+    sequence_number: str | None = Field(
+        None,
+        alias="aws:SequenceNumber",
+        description="Sequence number of the event in the bus.",
+    )
+    delivery_type: Literal["LIVE", "REPLAY"] = Field(
+        alias="aws:DeliveryType",
+        description="Whether the event is delivered live or as part of a replay.",
+    )
+    ingestion_time: str | None = Field(None, alias="aws:IngestionTime", description="Time the bus ingested the event.")
+    event_group_id: str | None = Field(None, alias="EventGroupId", description="FIFO group id of the event.")
+    deduplication_id: str | None = Field(None, alias="DeduplicationId", description="Deduplication id of the event.")
+    content_type: str | None = Field(
+        None,
+        alias="ContentType",
+        description="Content type of the payload.",
+        examples=["application/json", "application/eventbridge+json", "application/octet-stream"],
+    )
+    source: str | None = Field(None, alias="aws:Source", description="Event source, only for PutEvents.")
+    detail_type: str | None = Field(None, alias="aws:DetailType", description="Detail type, only for PutEvents.")
+
+
+class EventBridgeV2RecordModel(BaseModel):
+    data: Any = Field(
+        alias="Data",
+        description="Event payload. Bytes for `application/octet-stream`, the classic envelope for PutEvents.",
+    )
+    metadata: dict[str, Any] = Field(default_factory=dict, alias="Metadata", description="Producer metadata.")
+    system_metadata: EventBridgeV2SystemMetadataModel = Field(
+        alias="SystemMetadata",
+        description="Metadata added by the bus.",
+    )
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    @model_validator(mode="before")
+    @classmethod
+    def decode_octet_stream(cls, values: Any) -> Any:
+        if not isinstance(values, dict):
+            return values
+        data = values.get("Data", values.get("data"))
+        system_metadata = values.get("SystemMetadata", values.get("system_metadata")) or {}
+        content_type = system_metadata.get("ContentType") if isinstance(system_metadata, dict) else None
+        if content_type == OCTET_STREAM and isinstance(data, str):
+            return {**values, "Data": base64.b64decode(data)}
+        return values
+
+
+class EventBridgeV2Model(RootModel[list[EventBridgeV2RecordModel]]):
+    """Amazon EventBridge Custom Event Bus delivery using the WITH_METADATA transformer.
+
+    Accepts a single object or an array, and always exposes a list of records in `root`.
+    """
+
+    @model_validator(mode="before")
+    @classmethod
+    def wrap_single_record(cls, value: Any) -> Any:
+        return value if isinstance(value, list) else [value]

docs/utilities/data_classes.md

@@ -87,6 +87,7 @@ Each event source is linked to its corresponding GitHub file with the full set o
 | [Connect Contact Flow](#connect-contact-flow) | `ConnectContactFlowEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/connect_contact_flow_event.py) |
 | [DynamoDB streams](#dynamodb-streams) | `DynamoDBStreamEvent`, `DynamoDBRecordEventName` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py) |
 | [EventBridge](#eventbridge) | `EventBridgeEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/event_bridge_event.py) |
+| [EventBridge Custom Event Bus](#eventbridge-custom-event-bus) | `EventBridgeV2Event` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/event_bridge_v2_event.py) |
 | [Kafka](#kafka) | `KafkaEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/kafka_event.py) |
 | [Kinesis Data Stream](#kinesis-streams) | `KinesisStreamEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py) |
 | [Kinesis Firehose Delivery Stream](#kinesis-firehose-delivery-stream) | `KinesisFirehoseEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/kinesis_firehose_event.py) |
@@ -552,6 +553,23 @@ The class automatically deserializes DynamoDB types into their equivalent Python
     --8<-- "tests/events/eventBridgeEvent.json"
     ```
 
+### EventBridge Custom Event Bus
+
+The [Custom Event Bus](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-custom-bus.html) delivers a single object or an array of events, depending on the subscriber batching. `EventBridgeV2Event` always exposes them through `records`.
+
+Use the `WITH_METADATA` transformer when you need event ids, FIFO groups or replay detection (`system_metadata.delivery_type`). `RAW` deliveries carry no marker, so records are still accepted (classic envelope, PutEvents array or bare PutRawEvents payload) but `metadata` and `system_metadata` are empty.
+
+=== "app.py"
+
+    ```python hl_lines="1 5"
+    --8<-- "examples/event_sources/src/eventBridgeV2Event.py"
+    ```
+
+=== "EventBridge Custom Event Bus Example Event"
+    ```json
+    --8<-- "tests/events/eventBridgeV2Event.json"
+    ```
+
 ### Kafka
 
 This example is based on the AWS docs for [Amazon MSK](https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html){target="_blank"} and [self-managed Apache Kafka](https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html){target="_blank"}.

docs/utilities/parser.md

@@ -135,6 +135,7 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **DynamoDBStreamModel**                           | Lambda Event Source payload for Amazon DynamoDB Streams                                       |
 | **DynamoDBStreamLambdaOnFailureDestinationModel** | Lambda on-failure destination payload for Amazon DynamoDB Streams                                          |
 | **EventBridgeModel**                              | Lambda Event Source payload for Amazon EventBridge                                            |
+| **EventBridgeV2Model**                            | Lambda Event Source payload for Amazon EventBridge Custom Event Bus (`WITH_METADATA`, single or array)  |
 | **IoTCoreThingEvent**                             | Lambda Event Source payload for IoT Core Thing created, updated, or deleted.                  |
 | **IoTCoreThingTypeEvent**                         | Lambda Event Source payload for IoT Core Thing Type events.                                   |
 | **IoTCoreThingTypeAssociationEvent**              | Lambda Event Source payload for IoT Core Thing Type associated or disassociated with a Thing. |
@@ -209,6 +210,7 @@ You can use pre-built envelopes provided by the Parser to extract and parse spec
 | ----------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------- |
 | **DynamoDBStreamEnvelope**    | 1. Parses data using `DynamoDBStreamModel`. `` 2. Parses records in `NewImage` and `OldImage` keys using your model. `` 3. Returns a list with a dictionary containing `NewImage` and `OldImage` keys | `List[Dict[str, Optional[Model]]]` |
 | **EventBridgeEnvelope**       | 1. Parses data using `EventBridgeModel`. ``2. Parses `detail` key using your model`` and returns it.                                                                                                     | `Model`                            |
+| **EventBridgeV2Envelope**     | 1. Parses data using `EventBridgeV2Model`. ``2. Parses `Data` (or `Data.detail` for PutEvents) of each record using your model`` and returns them in a list. Also accepts RAW deliveries. | `List[Model]`                      |
 | **SqsEnvelope**               | 1. Parses data using `SqsModel`. ``2. Parses records in `body` key using your model`` and return them in a list.                                                                                         | `List[Model]`                      |
 | **CloudWatchLogsEnvelope**    | 1. Parses data using `CloudwatchLogsModel` which will base64 decode and decompress it. ``2. Parses records in `message` key using your model`` and return them in a list.                                | `List[Model]`                      |
 | **KinesisDataStreamEnvelope** | 1. Parses data using `KinesisDataStreamModel` which will base64 decode it. ``2. Parses records in the `Records` key using your model`` and returns them in a list.                                       | `List[Model]`                      |

examples/event_sources/src/eventBridgeV2Event.py

@@ -0,0 +1,11 @@
+from aws_lambda_powertools.utilities.data_classes import EventBridgeV2Event, event_source
+
+
+@event_source(data_class=EventBridgeV2Event)
+def lambda_handler(event: EventBridgeV2Event, context):
+    for record in event.records:
+        if record.system_metadata.delivery_type == "REPLAY":
+            continue
+
+        # Do something with record.data, or record.detail when published with PutEvents
+        print(record.system_metadata.event_id, record.data)