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) ‑> Any

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

Returns

Dict
The validated event. If the schema specifies a default value for fields that are omitted, those default values will be included in the 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
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