chore: add rabbitmq data_class

Author: Michael Brewer

aws_lambda_powertools/utilities/data_classes/rabbit_mq_event.py

@@ -0,0 +1,120 @@
+import base64
+import json
+from typing import Any, Dict, List
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class BasicProperties(DictWrapper):
+    @property
+    def content_type(self) -> str:
+        return self["contentType"]
+
+    @property
+    def content_encoding(self) -> str:
+        return self["contentEncoding"]
+
+    @property
+    def headers(self) -> Dict[str, Any]:
+        return self["headers"]
+
+    @property
+    def delivery_mode(self) -> int:
+        return self["deliveryMode"]
+
+    @property
+    def priority(self) -> int:
+        return self["priority"]
+
+    @property
+    def correlation_id(self) -> str:
+        return self["correlationId"]
+
+    @property
+    def reply_to(self) -> str:
+        return self["replyTo"]
+
+    @property
+    def expiration(self) -> int:
+        return self["expiration"]
+
+    @property
+    def message_id(self) -> str:
+        return self["messageId"]
+
+    @property
+    def timestamp(self) -> int:
+        return self["timestamp"]
+
+    @property
+    def get_type(self) -> str:
+        return self["type"]
+
+    @property
+    def user_id(self) -> str:
+        return self["userId"]
+
+    @property
+    def app_id(self) -> str:
+        return self["appId"]
+
+    @property
+    def cluster_id(self) -> str:
+        return self["clusterId"]
+
+    @property
+    def body_size(self) -> int:
+        return self["bodySize"]
+
+
+class RabbitMessage(DictWrapper):
+    @property
+    def basic_properties(self) -> BasicProperties:
+        return BasicProperties(self["basicProperties"])
+
+    @property
+    def redelivered(self) -> bool:
+        return self["redelivered"]
+
+    @property
+    def data(self) -> str:
+        return self["data"]
+
+    @property
+    def decoded_data(self) -> str:
+        """Decodes the data as a str"""
+        return base64.b64decode(self.data.encode()).decode()
+
+    @property
+    def json_data(self) -> Any:
+        """Parses the data as json"""
+        return json.loads(self.decoded_data)
+
+
+class RabbitMQEvent(DictWrapper):
+    """Represents a Rabbit MQ event sent to Lambda
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/lambda/latest/dg/with-mq.html
+    - https://aws.amazon.com/blogs/compute/using-amazon-mq-for-rabbitmq-as-an-event-source-for-lambda/
+    """
+
+    def __init__(self, data: Dict[str, Any]):
+        super().__init__(data)
+        self._rmq_messages_by_queue = {
+            key: [RabbitMessage(message) for message in messages]
+            for key, messages in self["rmqMessagesByQueue"].items()
+        }
+
+    @property
+    def event_source(self) -> str:
+        return self["eventSource"]
+
+    @property
+    def event_source_arn(self) -> str:
+        return self["eventSourceArn"]
+
+    @property
+    def rmq_messages_by_queue(self) -> Dict[str, List[RabbitMessage]]:
+        return self._rmq_messages_by_queue

tests/functional/data_classes/test_amazon_mq.py

@@ -1,8 +1,11 @@
+from typing import Dict
+
 from aws_lambda_powertools.utilities.data_classes.active_mq_event import ActiveMQEvent, ActiveMQMessage
+from aws_lambda_powertools.utilities.data_classes.rabbit_mq_event import BasicProperties, RabbitMessage, RabbitMQEvent
 from tests.functional.utils import load_event
 
 
-def test_activate_mq_event():
+def test_active_mq_event():
     event = ActiveMQEvent(load_event("activeMQEvent.json"))
 
     assert event.event_source == "aws:amq"
@@ -31,3 +34,27 @@ def test_activate_mq_event():
     messages = list(event.messages)
     message = messages[1]
     assert message.json_data["timeout"] == 0
+
+
+def test_rabbit_mq_event():
+    event = RabbitMQEvent(load_event("rabbitMQEvent.json"))
+
+    assert event.event_source == "aws:rmq"
+    assert event.event_source_arn is not None
+
+    message = event.rmq_messages_by_queue["pizzaQueue::/"][0]
+    assert isinstance(message, RabbitMessage)
+    properties = message.basic_properties
+    assert isinstance(properties, BasicProperties)
+    assert properties.content_type == "text/plain"
+    assert properties.content_encoding is None
+    assert isinstance(properties.headers, Dict)
+    assert properties.headers["header1"] is not None
+    assert properties.delivery_mode == 1
+    assert properties.priority == 34
+    assert properties.correlation_id is None
+
+    assert message.redelivered is False
+    assert message.data is not None
+    assert message.decoded_data is not None
+    assert message.json_data["timeout"] == 0