Module `aws_lambda_powertools.utilities.validation`

Simple validator to enforce incoming/outgoing event conforms with JSON Schema

Sub-modules

aws_lambda_powertools.utilities.validation.base
aws_lambda_powertools.utilities.validation.envelopes: Built-in envelopes
aws_lambda_powertools.utilities.validation.exceptions

Functions

def validate(event: Any, schema: Dict, formats: Optional[Dict] = None, handlers: Optional[Dict] = None, provider_options: Optional[Dict] = None, envelope: Optional[str] = None, jmespath_options: Optional[Dict] = None)

Standalone function to validate event data using a JSON Schema

Typically used when you need more control over the validation process.

Parameters

event : Dict: Lambda event to be validated
schema : Dict: JSON Schema to validate incoming event
envelope : Dict: JMESPath expression to filter data against
jmespath_options : Dict: Alternative JMESPath options to be included when filtering expr
formats : Dict: Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool
handlers : Dict: Custom methods to retrieve remote schemes, keyed off of URI scheme
provider_options : Dict: Arguments that will be passed directly to the underlying validate call

Example

Validate event

from aws_lambda_powertools.utilities.validation import validate

def handler(event, context):
    validate(event=event, schema=json_schema_dict)
    return event

Unwrap event before validating against actual payload - using built-in envelopes

from aws_lambda_powertools.utilities.validation import validate, envelopes

def handler(event, context):
    validate(event=event, schema=json_schema_dict, envelope=envelopes.API_GATEWAY_REST)
    return event

Unwrap event before validating against actual payload - using custom JMESPath expression

from aws_lambda_powertools.utilities.validation import validate

def handler(event, context):
    validate(event=event, schema=json_schema_dict, envelope="payload[*].my_data")
    return event

Unwrap and deserialize JSON string event before validating against actual payload - using built-in functions

from aws_lambda_powertools.utilities.validation import validate

def handler(event, context):
    validate(event=event, schema=json_schema_dict, envelope="Records[*].powertools_json(body)")
    return event

Unwrap, decode base64 and deserialize JSON string event before validating against actual payload - using built-in functions

from aws_lambda_powertools.utilities.validation import validate

def handler(event, context):
    validate(event=event, schema=json_schema_dict, envelope="Records[*].kinesis.powertools_json(powertools_base64(data))")
    return event

Unwrap, decompress ZIP archive and deserialize JSON string event before validating against actual payload - using built-in functions

from aws_lambda_powertools.utilities.validation import validate

def handler(event, context):
    validate(event=event, schema=json_schema_dict, envelope="awslogs.powertools_base64_gzip(data) | powertools_json(@).logEvents[*]")
    return event

Raises

SchemaValidationError: When schema validation fails against data set
InvalidSchemaFormatError: When JSON schema provided is invalid
InvalidEnvelopeExpressionError: When JMESPath expression to unwrap event is invalid

def validator(handler: Callable, event: Union[Dict, str], context: Any, inbound_schema: Optional[Dict] = None, inbound_formats: Optional[Dict] = None, inbound_handlers: Optional[Dict] = None, inbound_provider_options: Optional[Dict] = None, outbound_schema: Optional[Dict] = None, outbound_formats: Optional[Dict] = None, outbound_handlers: Optional[Dict] = None, outbound_provider_options: Optional[Dict] = None, envelope: str = '', jmespath_options: Optional[Dict] = None, **kwargs: Any) ‑> Any

Lambda handler decorator to validate incoming/outbound data using a JSON Schema

Parameters

handler : Callable: Method to annotate on
event : Dict: Lambda event to be validated
context : Any: Lambda context object
inbound_schema : Dict: JSON Schema to validate incoming event
outbound_schema : Dict: JSON Schema to validate outbound event
envelope : Dict: JMESPath expression to filter data against
jmespath_options : Dict: Alternative JMESPath options to be included when filtering expr
inbound_formats : Dict: Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool
outbound_formats : Dict: Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool
inbound_handlers : Dict: Custom methods to retrieve remote schemes, keyed off of URI scheme
outbound_handlers : Dict: Custom methods to retrieve remote schemes, keyed off of URI scheme
inbound_provider_options : Dict: Arguments that will be passed directly to the underlying validation call, in this case fastjsonchema.validate. For all supported arguments see: https://horejsek.github.io/python-fastjsonschema/#fastjsonschema.validate
outbound_provider_options : Dict: Arguments that will be passed directly to the underlying validation call, in this case fastjsonchema.validate. For all supported arguments see: https://horejsek.github.io/python-fastjsonschema/#fastjsonschema.validate

Example

Validate incoming event

from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=json_schema_dict)
def handler(event, context):
    return event

Validate incoming and outgoing event

from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=json_schema_dict, outbound_schema=response_json_schema_dict)
def handler(event, context):
    return event

Unwrap event before validating against actual payload - using built-in envelopes

from aws_lambda_powertools.utilities.validation import validator, envelopes

@validator(inbound_schema=json_schema_dict, envelope=envelopes.API_GATEWAY_REST)
def handler(event, context):
    return event

Unwrap event before validating against actual payload - using custom JMESPath expression

from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=json_schema_dict, envelope="payload[*].my_data")
def handler(event, context):
    return event

Unwrap and deserialize JSON string event before validating against actual payload - using built-in functions

from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=json_schema_dict, envelope="Records[*].powertools_json(body)")
def handler(event, context):
    return event

Unwrap, decode base64 and deserialize JSON string event before validating against actual payload - using built-in functions

from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=json_schema_dict, envelope="Records[*].kinesis.powertools_json(powertools_base64(data))")
def handler(event, context):
    return event

Unwrap, decompress ZIP archive and deserialize JSON string event before validating against actual payload - using built-in functions

from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=json_schema_dict, envelope="awslogs.powertools_base64_gzip(data) | powertools_json(@).logEvents[*]")
def handler(event, context):
    return event

Returns

Any: Lambda handler response

Raises

SchemaValidationError: When schema validation fails against data set
InvalidSchemaFormatError: When JSON schema provided is invalid
InvalidEnvelopeExpressionError: When JMESPath expression to unwrap event is invalid

Classes

class InvalidEnvelopeExpressionError (*args, **kwargs)

When JMESPath fails to parse expression

Expand source code

class InvalidEnvelopeExpressionError(Exception):
    """When JMESPath fails to parse expression"""

Ancestors

builtins.Exception
builtins.BaseException

class InvalidSchemaFormatError (*args, **kwargs)

When JSON Schema is in invalid format

Expand source code

class InvalidSchemaFormatError(Exception):
    """When JSON Schema is in invalid format"""

Ancestors

builtins.Exception
builtins.BaseException

class SchemaValidationError (message: Optional[str] = None, validation_message: Optional[str] = None, name: Optional[str] = None, path: Optional[List] = None, value: Optional[Any] = None, definition: Optional[Any] = None, rule: Optional[str] = None, rule_definition: Optional[Any] = None)

When serialization fail schema validation

Parameters

message : str, optional: Powertools for AWS Lambda (Python) formatted error message
validation_message : str, optional: Containing human-readable information what is wrong (e.g. data.property[index] must be smaller than or equal to 42)
name : str, optional: name of a path in the data structure (e.g. data.property[index])
path : List, optional: path as an array in the data structure (e.g. ['data', 'property', 'index']),
value : Any, optional: The invalid value
definition : Any, optional: The full rule definition (e.g. 42)
rule : str, optional: rule which the data is breaking (e.g. maximum)
rule_definition : Any, optional: The specific rule definition (e.g. 42)

Expand source code

class SchemaValidationError(Exception):
    """When serialization fail schema validation"""

    def __init__(
        self,
        message: Optional[str] = None,
        validation_message: Optional[str] = None,
        name: Optional[str] = None,
        path: Optional[List] = None,
        value: Optional[Any] = None,
        definition: Optional[Any] = None,
        rule: Optional[str] = None,
        rule_definition: Optional[Any] = None,
    ):
        """

        Parameters
        ----------
        message : str, optional
            Powertools for AWS Lambda (Python) formatted error message
        validation_message : str, optional
            Containing human-readable information what is wrong
            (e.g. `data.property[index] must be smaller than or equal to 42`)
        name : str, optional
            name of a path in the data structure
            (e.g. `data.property[index]`)
        path: List, optional
            `path` as an array in the data structure
            (e.g. `['data', 'property', 'index']`),
        value : Any, optional
            The invalid value
        definition : Any, optional
            The full rule `definition`
            (e.g. `42`)
        rule : str, optional
            `rule` which the `data` is breaking
            (e.g. `maximum`)
        rule_definition : Any, optional
            The specific rule `definition`
            (e.g. `42`)
        """
        super().__init__(message)
        self.message = message
        self.validation_message = validation_message
        self.name = name
        self.path = path
        self.value = value
        self.definition = definition
        self.rule = rule
        self.rule_definition = rule_definition

Ancestors

builtins.Exception
builtins.BaseException