Validation

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

Usage Documentation

Validation

MODULE	DESCRIPTION
base
envelopes	Built-in envelopes
exceptions
validator

CLASS	DESCRIPTION
InvalidEnvelopeExpressionError	When JMESPath fails to parse expression
InvalidSchemaFormatError	When JSON Schema is in invalid format
SchemaValidationError	When serialization fail schema validation

FUNCTION	DESCRIPTION
validate	Standalone function to validate event data using a JSON Schema

InvalidEnvelopeExpressionError

Bases: Exception

When JMESPath fails to parse expression

InvalidSchemaFormatError

Bases: Exception

When JSON Schema is in invalid format

SchemaValidationError

SchemaValidationError(
    message: str | None = None,
    validation_message: str | None = None,
    name: str | None = None,
    path: list | None = None,
    value: Any | None = None,
    definition: Any | None = None,
    rule: str | None = None,
    rule_definition: Any | None = None,
)

Bases: Exception

When serialization fail schema validation

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

Source code in aws_lambda_powertools/utilities/validation/exceptions.py

def __init__(
    self,
    message: str | None = None,
    validation_message: str | None = None,
    name: str | None = None,
    path: list | None = None,
    value: Any | None = None,
    definition: Any | None = None,
    rule: str | None = None,
    rule_definition: Any | None = 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`)
    """
    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

validate

validate(
    event: Any,
    schema: dict,
    formats: dict | None = None,
    handlers: dict | None = None,
    provider_options: dict | None = None,
    envelope: str | None = None,
    jmespath_options: dict | None = None,
) -> Any

Standalone function to validate event data using a JSON Schema

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

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

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	DESCRIPTION
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	DESCRIPTION
SchemaValidationError	When schema validation fails against data set
InvalidSchemaFormatError	When JSON schema provided is invalid
InvalidEnvelopeExpressionError	When JMESPath expression to unwrap event is invalid

Source code in aws_lambda_powertools/utilities/validation/validator.py

def validate(
    event: Any,
    schema: dict,
    formats: dict | None = None,
    handlers: dict | None = None,
    provider_options: dict | None = None,
    envelope: str | None = None,
    jmespath_options: dict | None = 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
    """  # noqa: E501
    if envelope:
        event = jmespath_utils.query(
            data=event,
            envelope=envelope,
            jmespath_options=jmespath_options,
        )

    return validate_data_against_schema(
        data=event,
        schema=schema,
        formats=formats,
        handlers=handlers,
        provider_options=provider_options,
    )

base

FUNCTION	DESCRIPTION
validate_data_against_schema	Validate dict data against given JSON Schema

validate_data_against_schema

validate_data_against_schema(
    data: dict | str,
    schema: dict,
    formats: dict | None = None,
    handlers: dict | None = None,
    provider_options: dict | None = None,
) -> dict | str

Validate dict data against given JSON Schema

PARAMETER	DESCRIPTION
data	Data set to be validated TYPE: dict
schema	JSON Schema to validate against TYPE: dict
formats	Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool TYPE: dict | None DEFAULT: None
handlers	Custom methods to retrieve remote schemes, keyed off of URI scheme TYPE: dict | None DEFAULT: None
provider_options	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 TYPE: dict | None DEFAULT: None

RETURNS	DESCRIPTION
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	DESCRIPTION
SchemaValidationError	When schema validation fails against data set
InvalidSchemaFormatError	When JSON schema provided is invalid

Source code in aws_lambda_powertools/utilities/validation/base.py

def validate_data_against_schema(
    data: dict | str,
    schema: dict,
    formats: dict | None = None,
    handlers: dict | None = None,
    provider_options: dict | None = None,
) -> dict | str:
    """Validate dict data against given JSON Schema

    Parameters
    ----------
    data : dict
        Data set to be validated
    schema : dict
        JSON Schema to validate against
    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 validation call, in this case fastjsonchema.validate.
        For all supported arguments see: https://horejsek.github.io/python-fastjsonschema/#fastjsonschema.validate

    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
    """
    try:
        formats = formats or {}
        handlers = handlers or {}
        provider_options = provider_options or {}
        return fastjsonschema.validate(
            definition=schema,
            data=data,
            formats=formats,
            handlers=handlers,
            **provider_options,
        )
    except (TypeError, AttributeError, fastjsonschema.JsonSchemaDefinitionException) as e:
        raise InvalidSchemaFormatError(f"Schema received: {schema}, Formats: {formats}. Error: {e}")
    except fastjsonschema.JsonSchemaValueException as e:
        message = f"Failed schema validation. Error: {e.message}, Path: {e.path}, Data: {e.value}"  # noqa: B306
        raise SchemaValidationError(
            message,
            validation_message=e.message,  # noqa: B306
            name=e.name,
            path=e.path,
            value=e.value,
            definition=e.definition,
            rule=e.rule,
            rule_definition=e.rule_definition,
        )

envelopes

Built-in envelopes

exceptions

CLASS	DESCRIPTION
InvalidEnvelopeExpressionError	When JMESPath fails to parse expression
InvalidSchemaFormatError	When JSON Schema is in invalid format
SchemaValidationError	When serialization fail schema validation

InvalidEnvelopeExpressionError

Bases: Exception

When JMESPath fails to parse expression

InvalidSchemaFormatError

Bases: Exception

When JSON Schema is in invalid format

SchemaValidationError

SchemaValidationError(
    message: str | None = None,
    validation_message: str | None = None,
    name: str | None = None,
    path: list | None = None,
    value: Any | None = None,
    definition: Any | None = None,
    rule: str | None = None,
    rule_definition: Any | None = None,
)

Bases: Exception

When serialization fail schema validation

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

Source code in aws_lambda_powertools/utilities/validation/exceptions.py

def __init__(
    self,
    message: str | None = None,
    validation_message: str | None = None,
    name: str | None = None,
    path: list | None = None,
    value: Any | None = None,
    definition: Any | None = None,
    rule: str | None = None,
    rule_definition: Any | None = 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`)
    """
    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

validator

FUNCTION	DESCRIPTION
validate	Standalone function to validate event data using a JSON Schema
validator	Lambda handler decorator to validate incoming/outbound data using a JSON Schema

validate

validate(
    event: Any,
    schema: dict,
    formats: dict | None = None,
    handlers: dict | None = None,
    provider_options: dict | None = None,
    envelope: str | None = None,
    jmespath_options: dict | None = None,
) -> Any

Standalone function to validate event data using a JSON Schema

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

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

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	DESCRIPTION
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	DESCRIPTION
SchemaValidationError	When schema validation fails against data set
InvalidSchemaFormatError	When JSON schema provided is invalid
InvalidEnvelopeExpressionError	When JMESPath expression to unwrap event is invalid

Source code in aws_lambda_powertools/utilities/validation/validator.py

def validate(
    event: Any,
    schema: dict,
    formats: dict | None = None,
    handlers: dict | None = None,
    provider_options: dict | None = None,
    envelope: str | None = None,
    jmespath_options: dict | None = 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
    """  # noqa: E501
    if envelope:
        event = jmespath_utils.query(
            data=event,
            envelope=envelope,
            jmespath_options=jmespath_options,
        )

    return validate_data_against_schema(
        data=event,
        schema=schema,
        formats=formats,
        handlers=handlers,
        provider_options=provider_options,
    )

validator

validator(
    handler: Callable,
    event: dict | str,
    context: Any,
    inbound_schema: dict | None = None,
    inbound_formats: dict | None = None,
    inbound_handlers: dict | None = None,
    inbound_provider_options: dict | None = None,
    outbound_schema: dict | None = None,
    outbound_formats: dict | None = None,
    outbound_handlers: dict | None = None,
    outbound_provider_options: dict | None = None,
    envelope: str = "",
    jmespath_options: dict | None = None,
    **kwargs: Any
) -> Any

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

PARAMETER	DESCRIPTION
handler	Method to annotate on TYPE: Callable
event	Lambda event to be validated TYPE: dict
context	Lambda context object TYPE: Any
inbound_schema	JSON Schema to validate incoming event TYPE: dict DEFAULT: None
outbound_schema	JSON Schema to validate outbound event TYPE: dict DEFAULT: None
envelope	JMESPath expression to filter data against TYPE: dict DEFAULT: ''
jmespath_options	Alternative JMESPath options to be included when filtering expr TYPE: dict DEFAULT: None
inbound_formats	Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool TYPE: dict | None DEFAULT: None
outbound_formats	Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool TYPE: dict | None DEFAULT: None
inbound_handlers	Custom methods to retrieve remote schemes, keyed off of URI scheme TYPE: dict | None DEFAULT: None
outbound_handlers	Custom methods to retrieve remote schemes, keyed off of URI scheme TYPE: dict | None DEFAULT: None
inbound_provider_options	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 TYPE: dict | None DEFAULT: None
outbound_provider_options	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 TYPE: dict | None DEFAULT: None

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	DESCRIPTION
Any	Lambda handler response

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

Source code in aws_lambda_powertools/utilities/validation/validator.py

@lambda_handler_decorator
def validator(
    handler: Callable,
    event: dict | str,
    context: Any,
    inbound_schema: dict | None = None,
    inbound_formats: dict | None = None,
    inbound_handlers: dict | None = None,
    inbound_provider_options: dict | None = None,
    outbound_schema: dict | None = None,
    outbound_formats: dict | None = None,
    outbound_handlers: dict | None = None,
    outbound_provider_options: dict | None = None,
    envelope: str = "",
    jmespath_options: dict | None = 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
    """  # noqa: E501
    if envelope:
        event = jmespath_utils.query(
            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,
            formats=inbound_formats,
            handlers=inbound_handlers,
            provider_options=inbound_provider_options,
        )

    response = handler(event, context, **kwargs)

    if outbound_schema:
        logger.debug("Validating outbound event")
        validate_data_against_schema(
            data=response,
            schema=outbound_schema,
            formats=outbound_formats,
            handlers=outbound_handlers,
            provider_options=outbound_provider_options,
        )

    return response