Adding support for Kafka Consumer - first commit

Author: leandrodamascena

aws_lambda_powertools/utilities/kafka_consumer/__init__.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import Any
+
+from aws_lambda_powertools.utilities.data_classes.kafka_event import KafkaEventRecord
+
+
+class ConsumerRecord(KafkaEventRecord):
+    """
+    A Kafka Consumer Record
+    """
+
+    def __init__(self, data: dict[str, Any], json_deserializer=None):
+        super().__init__(data, json_deserializer=json_deserializer)
+        self._json_deserializer = json_deserializer

aws_lambda_powertools/utilities/kafka_consumer/consumer_record.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import io
+
+from avro.errors import SchemaResolutionException
+from avro.io import BinaryDecoder, DatumReader
+
+
+def deserialize_avro(avro_bytes, reader_schema: str | None = None):
+    """
+    Deserialize Avro binary data to Python objects
+
+    Parameters
+    ----------
+    avro_bytes: bytes
+        Avro binary data
+    reader_schema: str, Optional
+        Schema to use for reading
+
+    Returns
+    -------
+    dict
+        Deserialized Python object
+
+    Raises
+    ------
+    ValueError
+        If reader_schema schema is None or if deserialization fails
+    """
+    try:
+        reader = DatumReader(reader_schema)
+
+        decoder = BinaryDecoder(io.BytesIO(avro_bytes))
+        return reader.read(decoder)
+    except SchemaResolutionException as e:
+        raise ValueError(f"Schema mismatch: {e}") from e
+    except Exception as e:
+        raise ValueError(f"Failed to deserialize Avro data: {e}") from e

aws_lambda_powertools/utilities/kafka_consumer/functions.py

@@ -125,6 +125,8 @@ nox = "^2024.4.15"
 mkdocstrings-python = "^1.13.0"
 datadog-lambda = "^6.106.0"
 mkdocs-llmstxt = "^0.2.0"
+avro = "^1.12.0"
+protobuf = "^6.30.2"
 
 [tool.coverage.run]
 source = ["aws_lambda_powertools"]