feat: add powertools_base64 custom fn

Author: heitorlessa

aws_lambda_powertools/utilities/validation/base.py

@@ -6,7 +6,7 @@
 from jmespath.exceptions import LexerError
 
 from .exceptions import InvalidEnvelopeExpressionError, InvalidSchemaError, SchemaValidationError
-from .jmespath_functions import PowertoolsJson
+from .jmespath_functions import PowertoolsFunctions
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +23,7 @@ def validate_data_against_schema(data: Dict, schema: Dict):
 
 def unwrap_event_from_envelope(data: Dict, envelope: str, jmespath_options: Dict):
     if not jmespath_options:
-        jmespath_options = {"custom_functions": PowertoolsJson()}
+        jmespath_options = {"custom_functions": PowertoolsFunctions()}
 
     try:
         logger.debug(f"Envelope detected: {envelope}. JMESPath options: {jmespath_options}")

aws_lambda_powertools/utilities/validation/envelopes.py

@@ -1,8 +1,9 @@
 """Built-in envelopes"""
 
-API_GATEWAY_REST = "powertools(body)"
+API_GATEWAY_REST = "powertools_json(body)"
 API_GATEWAY_HTTP = API_GATEWAY_REST
 SQS = "Records[*].body"
 SNS = "Records[0].Sns.Message"
 EVENTBRIDGE = "detail"
 CLOUDWATCH_EVENTS_SCHEDULED = EVENTBRIDGE
+KINESIS_DATA_STREAM = "Records[*].kinesis.powertools_base64(data) | powertools_json(@)"

aws_lambda_powertools/utilities/validation/jmespath_functions.py

@@ -1,9 +1,14 @@
+import base64
 import json
 
 import jmespath
 
 
-class PowertoolsJson(jmespath.functions.Functions):
+class PowertoolsFunctions(jmespath.functions.Functions):
     @jmespath.functions.signature({"types": ["string"]})
     def _func_powertools_json(self, value):
         return json.loads(value)
+
+    @jmespath.functions.signature({"types": ["string"]})
+    def _func_powertools_base64(self, value):
+        return base64.b64decode(value).decode()