feature: Kafka Parser support

Author: Ran Isenberg

aws_lambda_powertools/utilities/parser/envelopes/kafka.py

@@ -0,0 +1,41 @@
+import logging
+from typing import Any, Dict, List, Optional, Type, Union
+
+from ..models import KafkaEventModel
+from ..types import Model
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class KafkaEnvelope(BaseEnvelope):
+    """Kafka event envelope to extract data within body key
+    The record's body parameter is a string, though it can also be a JSON encoded string.
+    Regardless of its type it'll be parsed into a BaseModel object.
+
+    Note: Records will be parsed the same way so if model is str,
+    all items in the list will be parsed as str and npt as JSON (and vice versa)
+    """
+
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]:
+        """Parses data found with model provided
+
+        Parameters
+        ----------
+        data : Dict
+            Lambda event to be parsed
+        model : Type[Model]
+            Data model provided to parse after extracting data using envelope
+
+        Returns
+        -------
+        Any
+            Parsed detail payload with model provided
+        """
+        logger.debug(f"Parsing incoming data with Kafka event model {KafkaEventModel}")
+        parsed_envelope: KafkaEventModel = KafkaEventModel.parse_obj(data)
+        logger.debug(f"Parsing Kafka event records in `value` with {model}")
+        ret_list = []
+        for records in parsed_envelope.records.values():
+            ret_list += [self._parse(data=record.value, model=model) for record in records]
+        return ret_list

aws_lambda_powertools/utilities/parser/models/kafka.py

@@ -0,0 +1,70 @@
+import base64
+import logging
+from binascii import Error as BinAsciiError
+from datetime import datetime
+from typing import Dict, List, Optional, Type, Union
+
+from pydantic import BaseModel, validator
+
+from aws_lambda_powertools.utilities.parser.types import Literal
+
+SERVERS_DELIMITER = ","
+
+logger = logging.getLogger(__name__)
+
+
+def _base64_decode(value: str) -> bytes:
+    try:
+        logger.debug("Decoding base64 Kafka record item before parsing")
+        return base64.b64decode(value)
+    except (BinAsciiError, TypeError):
+        raise ValueError("base64 decode failed")
+
+
+def _bytes_to_string(value: bytes) -> str:
+    try:
+        return value.decode("utf-8")
+    except (BinAsciiError, TypeError):
+        raise ValueError("base64 UTF-8 decode failed")
+
+
+class KafkaRecordModel(BaseModel):
+    topic: str
+    partition: int
+    offset: int
+    timestamp: datetime
+    timestampType: str
+    key: bytes
+    value: Union[str, Type[BaseModel]]
+    headers: List[Dict[str, bytes]]
+
+    # validators
+    _decode_key = validator("key", allow_reuse=True)(_base64_decode)
+
+    @validator("value", pre=True, allow_reuse=True)
+    def data_base64_decode(cls, value):
+        as_bytes = _base64_decode(value)
+        return _bytes_to_string(as_bytes)
+
+    @validator("headers", pre=True, allow_reuse=True)
+    def decode_headers_list(cls, value):
+        for header in value:
+            for key, values in header.items():
+                header[key] = bytes(values)
+        return value
+
+
+class KafkaEventModel(BaseModel):
+    """Self-managed Apache Kafka event trigger
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html
+    """
+
+    eventSource: Literal["aws:SelfManagedKafka"]
+    bootstrapServers: Optional[List[str]]
+    records: Dict[str, List[KafkaRecordModel]]
+
+    @validator("bootstrapServers", pre=True, allow_reuse=True)
+    def split_servers(cls, value):
+        return None if not value else value.split(SERVERS_DELIMITER)

tests/events/kafkaEvent.json

@@ -0,0 +1,34 @@
+{
+    "eventSource": "aws:SelfManagedKafka",
+    "bootstrapServers": "b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092,b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092",
+    "records": {
+        "mytopic-0": [
+            {
+                "topic": "mytopic",
+                "partition": 0,
+                "offset": 15,
+                "timestamp": 1545084650987,
+                "timestampType": "CREATE_TIME",
+                "key": "cmVjb3JkS2V5",
+                "value": "eyJrZXkiOiJ2YWx1ZSJ9",
+                "headers": [
+                    {
+                        "headerKey": [
+                            104,
+                            101,
+                            97,
+                            100,
+                            101,
+                            114,
+                            86,
+                            97,
+                            108,
+                            117,
+                            101
+                        ]
+                    }
+                ]
+            }
+        ]
+    }
+}

tests/functional/parser/test_kafka.py

@@ -0,0 +1,49 @@
+from typing import List
+
+from aws_lambda_powertools.utilities.parser import envelopes, event_parser
+from aws_lambda_powertools.utilities.parser.models import KafkaEventModel, KafkaRecordModel
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from tests.functional.parser.schemas import MyALambdaKafkaBusiness
+from tests.functional.utils import load_event
+
+
+@event_parser(model=MyALambdaKafkaBusiness, envelope=envelopes.KafkaEnvelope)
+def handle_lambda_kafka_with_envelope(event: List[MyALambdaKafkaBusiness], _: LambdaContext):
+    assert event[0].key == "value"
+    assert len(event) == 1
+
+
+@event_parser(model=KafkaEventModel)
+def handle_kafka_event(event: KafkaEventModel, _: LambdaContext):
+    return event
+
+
+def test_kafka_event_with_envelope():
+    event = load_event("kafkaEvent.json")
+    handle_lambda_kafka_with_envelope(event, LambdaContext())
+
+
+def test_kafka_event():
+    json_event = load_event("kafkaEvent.json")
+    event: KafkaEventModel = handle_kafka_event(json_event, LambdaContext())
+    assert event.eventSource == "aws:SelfManagedKafka"
+    bootstrap_servers = [
+        "b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092",
+        "b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092",
+    ]
+    assert event.bootstrapServers == bootstrap_servers
+
+    records = list(event.records["mytopic-0"])
+    assert len(records) == 1
+    record: KafkaRecordModel = records[0]
+    assert record.topic == "mytopic"
+    assert record.partition == 0
+    assert record.offset == 15
+    assert record.timestamp is not None
+    convert_time = int(round(record.timestamp.timestamp() * 1000))
+    assert convert_time == 1545084650987
+    assert record.timestampType == "CREATE_TIME"
+    assert record.key == b"recordKey"
+    assert record.value == '{"key":"value"}'
+    assert len(record.headers) == 1
+    assert record.headers[0]["headerKey"] == b"headerValue"