feat: simple JSON Schema validator utility

Makefile

@@ -4,7 +4,7 @@ target:
 
 dev:
 	pip install --upgrade pip poetry pre-commit
-	poetry install
+	poetry install --extras "jmespath"
 	pre-commit install
 
 dev-docs:

aws_lambda_powertools/utilities/validation/__init__.py

@@ -0,0 +1,14 @@
+"""
+Simple validator to enforce incoming/outgoing event conforms with JSON Schema
+"""
+
+from .exceptions import InvalidEnvelopeExpressionError, InvalidSchemaFormatError, SchemaValidationError
+from .validator import validate, validator
+
+__all__ = [
+    "validate",
+    "validator",
+    "InvalidSchemaFormatError",
+    "SchemaValidationError",
+    "InvalidEnvelopeExpressionError",
+]

aws_lambda_powertools/utilities/validation/base.py

@@ -0,0 +1,33 @@
+import logging
+from typing import Dict
+
+import fastjsonschema
+import jmespath
+from jmespath.exceptions import LexerError
+
+from .exceptions import InvalidEnvelopeExpressionError, InvalidSchemaFormatError, SchemaValidationError
+from .jmespath_functions import PowertoolsFunctions
+
+logger = logging.getLogger(__name__)
+
+
+def validate_data_against_schema(data: Dict, schema: Dict):
+    try:
+        fastjsonschema.validate(definition=schema, data=data)
+    except fastjsonschema.JsonSchemaException as e:
+        message = f"Failed schema validation. Error: {e.message}, Path: {e.path}, Data: {e.value}"  # noqa: B306, E501
+        raise SchemaValidationError(message)
+    except (TypeError, AttributeError) as e:
+        raise InvalidSchemaFormatError(f"Schema received: {schema}. Error: {e}")
+
+
+def unwrap_event_from_envelope(data: Dict, envelope: str, jmespath_options: Dict):
+    if not jmespath_options:
+        jmespath_options = {"custom_functions": PowertoolsFunctions()}
+
+    try:
+        logger.debug(f"Envelope detected: {envelope}. JMESPath options: {jmespath_options}")
+        return jmespath.search(envelope, data, options=jmespath.Options(**jmespath_options))
+    except (LexerError, TypeError, UnicodeError) as e:
+        message = f"Failed to unwrap event from envelope using expression. Error: {e} Exp: {envelope}, Data: {data}"  # noqa: B306, E501
+        raise InvalidEnvelopeExpressionError(message)

aws_lambda_powertools/utilities/validation/envelopes.py

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

aws_lambda_powertools/utilities/validation/exceptions.py

@@ -0,0 +1,14 @@
+class SchemaValidationError(Exception):
+    """When serialization fail schema validation"""
+
+    pass
+
+
+class InvalidSchemaFormatError(Exception):
+    """When JSON Schema is in invalid format"""
+
+    pass
+
+
+class InvalidEnvelopeExpressionError(Exception):
+    """When JMESPath fails to parse expression"""

aws_lambda_powertools/utilities/validation/jmespath_functions.py

@@ -0,0 +1,22 @@
+import base64
+import gzip
+import json
+
+import jmespath
+
+
+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()
+
+    @jmespath.functions.signature({"types": ["string"]})
+    def _func_powertools_base64_gzip(self, value):
+        encoded = base64.b64decode(value)
+        uncompressed = gzip.decompress(encoded)
+
+        return uncompressed.decode()

aws_lambda_powertools/utilities/validation/validator.py

@@ -0,0 +1,40 @@
+import logging
+from typing import Any, Callable, Dict, Union
+
+from ...middleware_factory import lambda_handler_decorator
+from .base import unwrap_event_from_envelope, validate_data_against_schema
+
+logger = logging.getLogger(__name__)
+
+
+@lambda_handler_decorator
+def validator(
+    handler: Callable,
+    event: Union[Dict, str],
+    context: Any,
+    inbound_schema: Dict = None,
+    outbound_schema: Dict = None,
+    envelope: str = None,
+    jmespath_options: Dict = None,
+):
+    if envelope:
+        event = unwrap_event_from_envelope(data=event, envelope=envelope, jmespath_options=jmespath_options)
+
+    if inbound_schema:
+        logger.debug("Validating inbound event")
+        validate_data_against_schema(data=event, schema=inbound_schema)
+
+    response = handler(event, context)
+
+    if outbound_schema:
+        logger.debug("Validating outbound event")
+        validate_data_against_schema(data=event, schema=outbound_schema)
+
+    return response
+
+
+def validate(event: Dict, schema: Dict = None, envelope: str = None, jmespath_options: Dict = None):
+    if envelope:
+        event = unwrap_event_from_envelope(data=event, envelope=envelope, jmespath_options=jmespath_options)
+
+    validate_data_against_schema(data=event, schema=schema)