diff --git a/aws_lambda_powertools/utilities/data_classes/__init__.py b/aws_lambda_powertools/utilities/data_classes/__init__.py index 655ad1a74ca..5fad544f3a1 100644 --- a/aws_lambda_powertools/utilities/data_classes/__init__.py +++ b/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", diff --git a/aws_lambda_powertools/utilities/data_classes/event_bridge_v2_event.py b/aws_lambda_powertools/utilities/data_classes/event_bridge_v2_event.py new file mode 100644 index 00000000000..a20bb05e734 --- /dev/null +++ b/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 + ] diff --git a/aws_lambda_powertools/utilities/parser/envelopes/__init__.py b/aws_lambda_powertools/utilities/parser/envelopes/__init__.py index 0bf4b7a5535..6a14cd12d72 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/__init__.py +++ b/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", diff --git a/aws_lambda_powertools/utilities/parser/envelopes/event_bridge_v2.py b/aws_lambda_powertools/utilities/parser/envelopes/event_bridge_v2.py new file mode 100644 index 00000000000..dddc2a19807 --- /dev/null +++ b/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] diff --git a/aws_lambda_powertools/utilities/parser/models/__init__.py b/aws_lambda_powertools/utilities/parser/models/__init__.py index 2dc2985d4c5..c164e67ae16 100644 --- a/aws_lambda_powertools/utilities/parser/models/__init__.py +++ b/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", diff --git a/aws_lambda_powertools/utilities/parser/models/event_bridge_v2.py b/aws_lambda_powertools/utilities/parser/models/event_bridge_v2.py new file mode 100644 index 00000000000..b9d6c047f20 --- /dev/null +++ b/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] diff --git a/docs/utilities/data_classes.md b/docs/utilities/data_classes.md index ca3ef69ac26..d57bdcd7509 100644 --- a/docs/utilities/data_classes.md +++ b/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"}. diff --git a/docs/utilities/parser.md b/docs/utilities/parser.md index a359d4364df..ecb8ede5f9f 100644 --- a/docs/utilities/parser.md +++ b/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]` | diff --git a/examples/event_sources/src/eventBridgeV2Event.py b/examples/event_sources/src/eventBridgeV2Event.py new file mode 100644 index 00000000000..818d6ac2a6c --- /dev/null +++ b/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) diff --git a/tests/events/eventBridgeV2Event.json b/tests/events/eventBridgeV2Event.json new file mode 100644 index 00000000000..9edfc2a3d26 --- /dev/null +++ b/tests/events/eventBridgeV2Event.json @@ -0,0 +1,40 @@ +[ + { + "Data": {"orderId": "s3-0", "total": 5}, + "Metadata": {"sub": "s3", "tenant": "acme"}, + "SystemMetadata": { + "EventGroupId": "43e6837b-3d96-44db-863d-e8abd80cd1e8", + "DeduplicationId": "43e6837b-3d96-44db-863d-e8abd80cd1e8", + "ContentType": "application/json", + "aws:IngestionTime": "2026-10-25T11:48:54.064Z", + "aws:EventId": "e3474ffa-0283-4fe7-af27-803e30cd92f4", + "aws:SequenceNumber": "10000000000000007000", + "aws:DeliveryType": "LIVE" + } + }, + { + "Data": { + "version": "0", + "id": "1cd1f343-9859-4e28-a539-89352846ce44", + "detail-type": "s6", + "source": "com.example.orders", + "account": "123456789012", + "time": "2026-10-25T11:48:53Z", + "region": "us-east-1", + "resources": [], + "detail": {"orderId": "s6-1", "total": 10} + }, + "Metadata": {}, + "SystemMetadata": { + "EventGroupId": "d8c85953-b07f-4bc1-b2ff-044a1c2b03bc", + "DeduplicationId": "d8c85953-b07f-4bc1-b2ff-044a1c2b03bc", + "ContentType": "application/eventbridge+json", + "aws:IngestionTime": "2026-10-25T11:48:53.977Z", + "aws:Source": "com.example.orders", + "aws:DetailType": "s6", + "aws:EventId": "b9c941f6-ccae-473e-95d5-1982a0b1e8a5", + "aws:SequenceNumber": "10000000000000006001", + "aws:DeliveryType": "REPLAY" + } + } +] diff --git a/tests/unit/data_classes/required_dependencies/test_event_bridge_v2_event.py b/tests/unit/data_classes/required_dependencies/test_event_bridge_v2_event.py new file mode 100644 index 00000000000..aee1f5c9a03 --- /dev/null +++ b/tests/unit/data_classes/required_dependencies/test_event_bridge_v2_event.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import base64 + +from aws_lambda_powertools.utilities.data_classes import EventBridgeV2Event +from tests.functional.utils import load_event + + +def test_event_bridge_v2_event_with_metadata_array(): + raw_event = load_event("eventBridgeV2Event.json") + event = EventBridgeV2Event(raw_event) + + raw_record, put_events_record = event.records + assert len(event.records) == 2 + + assert raw_record.data == raw_event[0]["Data"] + assert raw_record.detail is None + assert raw_record.metadata == {"sub": "s3", "tenant": "acme"} + assert raw_record.system_metadata.event_id == "e3474ffa-0283-4fe7-af27-803e30cd92f4" + assert raw_record.system_metadata.sequence_number == "10000000000000007000" + assert raw_record.system_metadata.delivery_type == "LIVE" + assert raw_record.system_metadata.event_group_id == "43e6837b-3d96-44db-863d-e8abd80cd1e8" + assert raw_record.system_metadata.deduplication_id == "43e6837b-3d96-44db-863d-e8abd80cd1e8" + assert raw_record.system_metadata.content_type == "application/json" + assert raw_record.system_metadata.ingestion_time == "2026-10-25T11:48:54.064Z" + assert raw_record.system_metadata.source is None + assert raw_record.system_metadata.detail_type is None + + assert put_events_record.detail == {"orderId": "s6-1", "total": 10} + assert put_events_record.metadata == {} + assert put_events_record.system_metadata.delivery_type == "REPLAY" + assert put_events_record.system_metadata.source == "com.example.orders" + assert put_events_record.system_metadata.detail_type == "s6" + + +def test_event_bridge_v2_event_single_object(): + raw_event = load_event("eventBridgeV2Event.json")[0] + event = EventBridgeV2Event(raw_event) + + assert len(event.records) == 1 + assert event.records[0].data == raw_event["Data"] + + +def test_event_bridge_v2_event_octet_stream(): + payload = b"\x00\x01binary" + raw_event = { + "Data": base64.b64encode(payload).decode(), + "Metadata": {}, + "SystemMetadata": {"ContentType": "application/octet-stream", "aws:EventId": "id"}, + } + + assert EventBridgeV2Event(raw_event).records[0].data == payload + + +def test_event_bridge_v2_event_raw_classic_envelope_and_array(): + classic = load_event("eventBridgeEvent.json") + + single = EventBridgeV2Event(classic).records + array = EventBridgeV2Event([classic, classic]).records + + assert len(single) == 1 + assert len(array) == 2 + assert single[0].data == classic + assert single[0].detail == classic["detail"] + assert single[0].metadata == {} + assert single[0].system_metadata.event_id is None + + +def test_event_bridge_v2_event_raw_put_raw_events_payload(): + event = EventBridgeV2Event({"orderId": "1", "total": 5}) + + assert event.records[0].data == {"orderId": "1", "total": 5} + assert event.records[0].detail is None diff --git a/tests/unit/parser/_pydantic/test_eventbridge_v2.py b/tests/unit/parser/_pydantic/test_eventbridge_v2.py new file mode 100644 index 00000000000..95c3a633ac3 --- /dev/null +++ b/tests/unit/parser/_pydantic/test_eventbridge_v2.py @@ -0,0 +1,104 @@ +import base64 + +import pytest +from pydantic import BaseModel + +from aws_lambda_powertools.utilities.parser import ValidationError, envelopes, parse +from aws_lambda_powertools.utilities.parser.models import EventBridgeV2Model +from tests.functional.utils import load_event + + +class Order(BaseModel): + orderId: str # noqa: N815 + total: int + + +def test_model_accepts_array_and_single(): + raw_event = load_event("eventBridgeV2Event.json") + + from_array = EventBridgeV2Model.model_validate(raw_event).root + from_single = EventBridgeV2Model.model_validate(raw_event[0]).root + + assert len(from_array) == 2 + assert len(from_single) == 1 + assert from_array[0].metadata == {"sub": "s3", "tenant": "acme"} + assert from_array[0].system_metadata.event_id == "e3474ffa-0283-4fe7-af27-803e30cd92f4" + assert from_array[0].system_metadata.delivery_type == "LIVE" + assert from_array[1].system_metadata.delivery_type == "REPLAY" + assert from_array[1].system_metadata.detail_type == "s6" + + +def test_model_decodes_octet_stream(): + raw_event = { + "Data": base64.b64encode(b"abc").decode(), + "Metadata": {}, + "SystemMetadata": { + "ContentType": "application/octet-stream", + "aws:EventId": "id", + "aws:SequenceNumber": "1", + "aws:DeliveryType": "LIVE", + "aws:IngestionTime": "2026-10-25T11:48:54.064Z", + }, + } + + assert EventBridgeV2Model.model_validate(raw_event).root[0].data == b"abc" + + +def test_model_rejects_invalid_delivery_type(): + raw_event = load_event("eventBridgeV2Event.json")[0] + raw_event["SystemMetadata"]["aws:DeliveryType"] = "OTHER" + + with pytest.raises(ValidationError): + EventBridgeV2Model.model_validate(raw_event) + + +def test_envelope_with_metadata_array_yields_data_and_detail(): + raw_event = load_event("eventBridgeV2Event.json") + + parsed = parse(event=raw_event, model=Order, envelope=envelopes.EventBridgeV2Envelope) + + assert parsed == [Order(orderId="s3-0", total=5), Order(orderId="s6-1", total=10)] + + +def test_envelope_with_metadata_single_object(): + raw_event = load_event("eventBridgeV2Event.json")[0] + + parsed = parse(event=raw_event, model=Order, envelope=envelopes.EventBridgeV2Envelope) + + assert parsed == [Order(orderId="s3-0", total=5)] + + +def test_envelope_raw_deliveries(): + classic = load_event("eventBridgeEvent.json") + classic["detail"] = {"orderId": "c", "total": 1} + + assert parse(event=classic, model=Order, envelope=envelopes.EventBridgeV2Envelope) == [Order(orderId="c", total=1)] + assert parse(event=[classic], model=Order, envelope=envelopes.EventBridgeV2Envelope) == [ + Order(orderId="c", total=1), + ] + assert parse(event={"orderId": "b", "total": 2}, model=Order, envelope=envelopes.EventBridgeV2Envelope) == [ + Order(orderId="b", total=2), + ] + + +def test_envelope_invalid_payload(): + raw_event = load_event("eventBridgeV2Event.json") + raw_event[0]["Data"] = {"orderId": "x"} + + with pytest.raises(ValidationError): + parse(event=raw_event, model=Order, envelope=envelopes.EventBridgeV2Envelope) + + +def test_envelope_parses_octet_stream_json_bytes(): + # GIVEN a JSON payload published as application/octet-stream, which the bus delivers base64 encoded + event = { + "Data": base64.b64encode(b'{"orderId": "o-1", "total": 5}').decode(), + "Metadata": {}, + "SystemMetadata": {"ContentType": "application/octet-stream", "aws:DeliveryType": "LIVE"}, + } + + # WHEN parsing with the envelope + orders = parse(event=event, model=Order, envelope=envelopes.EventBridgeV2Envelope) + + # THEN the bytes are decoded and parsed into the model, and the optional aws: fields may be absent + assert orders == [Order(orderId="o-1", total=5)]