Module aws_lambda_powertools.utilities.parser

Advanced event_parser utility

"""Advanced event_parser utility
"""
from . import envelopes
from .envelopes import BaseEnvelope
from .parser import event_parser, parse
from .pydantic import BaseModel, Field, ValidationError, root_validator, validator

__all__ = [
    "event_parser",
    "parse",
    "envelopes",
    "BaseEnvelope",
    "BaseModel",
    "Field",
    "validator",
    "root_validator",
    "ValidationError",
]

Sub-modules

aws_lambda_powertools.utilities.parser.envelopes

aws_lambda_powertools.utilities.parser.exceptions

aws_lambda_powertools.utilities.parser.models

aws_lambda_powertools.utilities.parser.parser

aws_lambda_powertools.utilities.parser.pydantic

aws_lambda_powertools.utilities.parser.types

Generics and other shared types used across parser

Functions

def Field(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: unicode = None, title: unicode = None, description: unicode = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), Any] = None, include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), Any] = None, const: bool = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, multiple_of: float = None, allow_inf_nan: bool = None, max_digits: int = None, decimal_places: int = None, min_items: int = None, max_items: int = None, unique_items: bool = None, min_length: int = None, max_length: int = None, allow_mutation: bool = True, regex: unicode = None, discriminator: unicode = None, repr: bool = True, **extra: Any) ‑> Any

Used to provide extra information about a field, either for the model schema or complex validation. Some arguments apply only to number fields (int, float, Decimal) and some apply only to str.

:param default: since this is replacing the field’s default, its first argument is used to set the default, use ellipsis (…) to indicate the field is required :param default_factory: callable that will be called when a default value is needed for this field If both default and default_factory are set, an error is raised. :param alias: the public name of the field :param title: can be any string, used in the schema :param description: can be any string, used in the schema :param exclude: exclude this field while dumping. Takes same values as the include and exclude arguments on the .dict method. :param include: include this field while dumping. Takes same values as the include and exclude arguments on the .dict method. :param const: this field is required and must take it's default value :param gt: only applies to numbers, requires the field to be "greater than". The schema will have an exclusiveMinimum validation keyword :param ge: only applies to numbers, requires the field to be "greater than or equal to". The schema will have a minimum validation keyword :param lt: only applies to numbers, requires the field to be "less than". The schema will have an exclusiveMaximum validation keyword :param le: only applies to numbers, requires the field to be "less than or equal to". The schema will have a maximum validation keyword :param multiple_of: only applies to numbers, requires the field to be "a multiple of". The schema will have a multipleOf validation keyword :param allow_inf_nan: only applies to numbers, allows the field to be NaN or infinity (+inf or -inf), which is a valid Python float. Default True, set to False for compatibility with JSON. :param max_digits: only applies to Decimals, requires the field to have a maximum number of digits within the decimal. It does not include a zero before the decimal point or trailing decimal zeroes. :param decimal_places: only applies to Decimals, requires the field to have at most a number of decimal places allowed. It does not include trailing decimal zeroes. :param min_items: only applies to lists, requires the field to have a minimum number of elements. The schema will have a minItems validation keyword :param max_items: only applies to lists, requires the field to have a maximum number of elements. The schema will have a maxItems validation keyword :param unique_items: only applies to lists, requires the field not to have duplicated elements. The schema will have a uniqueItems validation keyword :param min_length: only applies to strings, requires the field to have a minimum length. The schema will have a maximum validation keyword :param max_length: only applies to strings, requires the field to have a maximum length. The schema will have a maxLength validation keyword :param allow_mutation: a boolean which defaults to True. When False, the field raises a TypeError if the field is assigned on an instance. The BaseModel Config must set validate_assignment to True :param regex: only applies to strings, requires the field match against a regular expression pattern string. The schema will have a pattern validation keyword :param discriminator: only useful with a (discriminated a.k.a. tagged) Union of sub models with a common field. The discriminator is the name of this common field to shorten validation and improve generated schema :param repr: show this field in the representation :param **extra: any additional keyword arguments will be added as is to the schema

def event_parser(handler: Callable[[Any, aws_lambda_powertools.utilities.lambda_context.LambdaContext], ~EventParserReturnType], event: Dict[str, Any], context: LambdaContext, model: Type[~Model], envelope: Optional[Type[~Envelope]] = None) ‑> ~EventParserReturnType

Lambda handler decorator to parse & validate events using Pydantic models

It requires a model that implements Pydantic BaseModel to parse & validate the event.

When an envelope is given, it'll use the following logic:

1. Parse the event against the envelope model first e.g. EnvelopeModel(**event)
2. Envelope will extract a given key to be parsed against the model e.g. event.detail

This is useful when you need to confirm event wrapper structure, and b) selectively extract a portion of your payload for parsing & validation.

NOTE: If envelope is omitted, the complete event is parsed to match the model parameter BaseModel definition.

Example

Lambda handler decorator to parse & validate event

class Order(BaseModel):
    id: int
    description: str
    ...

@event_parser(model=Order)
def handler(event: Order, context: LambdaContext):
    ...

Lambda handler decorator to parse & validate event - using built-in envelope

class Order(BaseModel):
    id: int
    description: str
    ...

@event_parser(model=Order, envelope=envelopes.EVENTBRIDGE)
def handler(event: Order, context: LambdaContext):
    ...

Parameters

handler :  Callable

Method to annotate on

event :  Dict

Lambda event to be parsed & validated

context :  LambdaContext

Lambda context object

model :  Model

Your data model that will replace the event.

envelope : Envelope

Optional envelope to extract the model from

Raises

ValidationError

When input event does not conform with model provided

InvalidModelTypeError

When model given does not implement BaseModel

InvalidEnvelopeError

When envelope given does not implement BaseEnvelope

Expand source code

@lambda_handler_decorator
def event_parser(
    handler: Callable[[Any, LambdaContext], EventParserReturnType],
    event: Dict[str, Any],
    context: LambdaContext,
    model: Type[Model],
    envelope: Optional[Type[Envelope]] = None,
) -> EventParserReturnType:
    """Lambda handler decorator to parse & validate events using Pydantic models

    It requires a model that implements Pydantic BaseModel to parse & validate the event.

    When an envelope is given, it'll use the following logic:

    1. Parse the event against the envelope model first e.g. EnvelopeModel(**event)
    2. Envelope will extract a given key to be parsed against the model e.g. event.detail

    This is useful when you need to confirm event wrapper structure, and
    b) selectively extract a portion of your payload for parsing & validation.

    NOTE: If envelope is omitted, the complete event is parsed to match the model parameter BaseModel definition.

    Example
    -------
    **Lambda handler decorator to parse & validate event**

        class Order(BaseModel):
            id: int
            description: str
            ...

        @event_parser(model=Order)
        def handler(event: Order, context: LambdaContext):
            ...

    **Lambda handler decorator to parse & validate event - using built-in envelope**

        class Order(BaseModel):
            id: int
            description: str
            ...

        @event_parser(model=Order, envelope=envelopes.EVENTBRIDGE)
        def handler(event: Order, context: LambdaContext):
            ...

    Parameters
    ----------
    handler:  Callable
        Method to annotate on
    event:    Dict
        Lambda event to be parsed & validated
    context:  LambdaContext
        Lambda context object
    model:   Model
        Your data model that will replace the event.
    envelope: Envelope
        Optional envelope to extract the model from

    Raises
    ------
    ValidationError
        When input event does not conform with model provided
    InvalidModelTypeError
        When model given does not implement BaseModel
    InvalidEnvelopeError
        When envelope given does not implement BaseEnvelope
    """
    parsed_event = parse(event=event, model=model, envelope=envelope) if envelope else parse(event=event, model=model)
    logger.debug(f"Calling handler {handler.__name__}")
    return handler(parsed_event, context)

def parse(event: Dict[str, Any], model: Type[~Model], envelope: Optional[Type[~Envelope]] = None)

Standalone function to parse & validate events using Pydantic models

Typically used when you need fine-grained control over error handling compared to event_parser decorator.

Example

Lambda handler decorator to parse & validate event

from aws_lambda_powertools.utilities.parser import ValidationError

class Order(BaseModel):
    id: int
    description: str
    ...

def handler(event: Order, context: LambdaContext):
    try:
        parse(model=Order)
    except ValidationError:
        ...

Lambda handler decorator to parse & validate event - using built-in envelope

class Order(BaseModel):
    id: int
    description: str
    ...

def handler(event: Order, context: LambdaContext):
    try:
        parse(model=Order, envelope=envelopes.EVENTBRIDGE)
    except ValidationError:
        ...

Parameters

event :  Dict

Lambda event to be parsed & validated

model :  Model

Your data model that will replace the event

envelope : Envelope

Optional envelope to extract the model from

Raises

ValidationError

When input event does not conform with model provided

InvalidModelTypeError

When model given does not implement BaseModel

InvalidEnvelopeError

When envelope given does not implement BaseEnvelope

Expand source code

def parse(event: Dict[str, Any], model: Type[Model], envelope: Optional[Type[Envelope]] = None):
    """Standalone function to parse & validate events using Pydantic models

    Typically used when you need fine-grained control over error handling compared to event_parser decorator.

    Example
    -------

    **Lambda handler decorator to parse & validate event**

        from aws_lambda_powertools.utilities.parser import ValidationError

        class Order(BaseModel):
            id: int
            description: str
            ...

        def handler(event: Order, context: LambdaContext):
            try:
                parse(model=Order)
            except ValidationError:
                ...

    **Lambda handler decorator to parse & validate event - using built-in envelope**

        class Order(BaseModel):
            id: int
            description: str
            ...

        def handler(event: Order, context: LambdaContext):
            try:
                parse(model=Order, envelope=envelopes.EVENTBRIDGE)
            except ValidationError:
                ...

    Parameters
    ----------
    event:    Dict
        Lambda event to be parsed & validated
    model:   Model
        Your data model that will replace the event
    envelope: Envelope
        Optional envelope to extract the model from

    Raises
    ------
    ValidationError
        When input event does not conform with model provided
    InvalidModelTypeError
        When model given does not implement BaseModel
    InvalidEnvelopeError
        When envelope given does not implement BaseEnvelope
    """
    if envelope and callable(envelope):
        try:
            logger.debug(f"Parsing and validating event model with envelope={envelope}")
            return envelope().parse(data=event, model=model)
        except AttributeError:
            raise InvalidEnvelopeError(f"Envelope must implement BaseEnvelope, envelope={envelope}")

    try:
        logger.debug("Parsing and validating event model; no envelope used")
        if isinstance(event, str):
            return model.parse_raw(event)

        return model.parse_obj(event)
    except AttributeError:
        raise InvalidModelTypeError(f"Input model must implement BaseModel, model={model}")

def root_validator(*, pre: bool = False, allow_reuse: bool = False, skip_on_failure: bool = False) ‑> Union[AnyClassMethod, Callable[[Callable[..., Any]], AnyClassMethod]]

Decorate methods on a model indicating that they should be used to validate (and perhaps modify) data either before or after standard model parsing/validation is performed.

def validator(*fields: unicode, pre: bool = False, each_item: bool = False, always: bool = False, check_fields: bool = True, whole: bool = None, allow_reuse: bool = False) ‑> Callable[[Callable[..., Any]], AnyClassMethod]

Decorate methods on the class indicating that they should be used to validate fields :param fields: which field(s) the method should be called on :param pre: whether or not this validator should be called before the standard validators (else after) :param each_item: for complex objects (sets, lists etc.) whether to validate individual elements rather than the whole object :param always: whether this method and other validators should be called even if the value is missing :param check_fields: whether to check that the fields actually exist on the model :param allow_reuse: whether to track and raise an error if another validator refers to the decorated function

Classes

class BaseEnvelope

ABC implementation for creating a supported Envelope

Expand source code

class BaseEnvelope(ABC):
    """ABC implementation for creating a supported Envelope"""

    @staticmethod
    def _parse(data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Union[Model, None]:
        """Parses envelope data against model provided

        Parameters
        ----------
        data : Dict
            Data to be parsed and validated
        model : Type[Model]
            Data model to parse and validate data against

        Returns
        -------
        Any
            Parsed data
        """
        if data is None:
            logger.debug("Skipping parsing as event is None")
            return data

        logger.debug("parsing event against model")
        if isinstance(data, str):
            logger.debug("parsing event as string")
            return model.parse_raw(data)

        return model.parse_obj(data)

    @abstractmethod
    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]):
        """Implementation to parse data against envelope model, then against the data model

        NOTE: Call `_parse` method to fully parse data with model provided.

        Example
        -------

        **EventBridge envelope implementation example**

        def parse(...):
            # 1. parses data against envelope model
            parsed_envelope = EventBridgeModel(**data)

            # 2. parses portion of data within the envelope against model
            return self._parse(data=parsed_envelope.detail, model=data_model)
        """
        return NotImplemented  # pragma: no cover

Ancestors

• abc.ABC

Subclasses

• ApiGatewayEnvelope
• ApiGatewayV2Envelope
• CloudWatchLogsEnvelope
• DynamoDBStreamEnvelope
• EventBridgeEnvelope
• KafkaEnvelope
• KinesisDataStreamEnvelope
• KinesisFirehoseEnvelope
• LambdaFunctionUrlEnvelope
• SnsEnvelope
• SnsSqsEnvelope
• SqsEnvelope

Methods

def parse(self, data: Union[Dict[str, Any], Any, None], model: Type[~Model])

Implementation to parse data against envelope model, then against the data model

NOTE: Call _parse method to fully parse data with model provided.

Example

EventBridge envelope implementation example

def parse(…): # 1. parses data against envelope model parsed_envelope = EventBridgeModel(**data)

# 2. parses portion of data within the envelope against model
return self._parse(data=parsed_envelope.detail, model=data_model)

Expand source code

@abstractmethod
def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]):
    """Implementation to parse data against envelope model, then against the data model

    NOTE: Call `_parse` method to fully parse data with model provided.

    Example
    -------

    **EventBridge envelope implementation example**

    def parse(...):
        # 1. parses data against envelope model
        parsed_envelope = EventBridgeModel(**data)

        # 2. parses portion of data within the envelope against model
        return self._parse(data=parsed_envelope.detail, model=data_model)
    """
    return NotImplemented  # pragma: no cover

class BaseModel (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Ancestors

• pydantic.utils.Representation

Subclasses

• AlbModel
• AlbRequestContext
• AlbRequestContextData
• APIGatewayEventAuthorizer
• APIGatewayEventIdentity
• APIGatewayEventRequestContext
• APIGatewayProxyEventModel
• ApiGatewayUserCert
• ApiGatewayUserCertValidity
• APIGatewayProxyEventV2Model
• RequestContextV2
• RequestContextV2Authorizer
• RequestContextV2AuthorizerIam
• RequestContextV2AuthorizerIamCognito
• RequestContextV2AuthorizerJwt
• RequestContextV2Http
• CloudWatchLogsData
• CloudWatchLogsDecode
• CloudWatchLogsLogEvent
• CloudWatchLogsModel
• DynamoDBStreamChangedRecordModel
• DynamoDBStreamModel
• DynamoDBStreamRecordModel
• UserIdentity
• EventBridgeModel
• KafkaBaseEventModel
• KafkaRecordModel
• KinesisDataStreamModel
• KinesisDataStreamRecord
• KinesisDataStreamRecordPayload
• KinesisFirehoseModel
• KinesisFirehoseRecord
• KinesisFirehoseRecordMetadata
• S3Bucket
• S3EventRecordGlacierEventData
• S3EventRecordGlacierRestoreEventData
• S3Identity
• S3Message
• S3Model
• S3Object
• S3OwnerIdentify
• S3RecordModel
• S3RequestParameters
• S3ResponseElements
• S3ObjectConfiguration
• S3ObjectContext
• S3ObjectLambdaEvent
• S3ObjectSessionAttributes
• S3ObjectSessionContext
• S3ObjectSessionIssuer
• S3ObjectUserIdentity
• S3ObjectUserRequest
• SesMail
• SesMailCommonHeaders
• SesMailHeaders
• SesMessage
• SesModel
• SesReceipt
• SesReceiptAction
• SesReceiptVerdict
• SesRecordModel
• SnsModel
• SnsMsgAttributeModel
• SnsNotificationModel
• SnsRecordModel
• SqsAttributesModel
• SqsModel
• SqsMsgAttributeModel
• SqsRecordModel
• pydantic.env_settings.BaseSettings

Class variables

var Config

Static methods

def construct(**values: Any) ‑> Model

Creates a new model setting dict and fields_set from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = 'allow' was set since it adds all passed values

def from_orm(obj: Any) ‑> Model

def parse_file(path: Union[str, pathlib.Path], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: pydantic.parse.Protocol = None, allow_pickle: bool = False) ‑> Model

def parse_obj(obj: Any) ‑> Model

def parse_raw(b: Union[str, bytes], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: pydantic.parse.Protocol = None, allow_pickle: bool = False) ‑> Model

def schema(by_alias: bool = True, ref_template: unicode = '#/definitions/{model}') ‑> DictStrAny

def schema_json(*, by_alias: bool = True, ref_template: unicode = '#/definitions/{model}', **dumps_kwargs: Any) ‑> unicode

def update_forward_refs(**localns: Any) ‑> None

Try to update ForwardRefs on fields based on this Model, globalns and localns.

def validate(value: Any) ‑> Model

Methods

def copy(self: Model, *, include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), None] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), None] = None, update: Optional[ForwardRef('DictStrAny')] = None, deep: bool = False) ‑> Model

Duplicate a model, optionally choose which fields to include, exclude and change.

:param include: fields to include in new model :param exclude: fields to exclude from new model, as with values this takes precedence over include :param update: values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data :param deep: set to True to make a deep copy of the model :return: new model instance

def dict(self, *, include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), None] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), None] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) ‑> DictStrAny

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

def json(self, *, include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), None] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), None] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) ‑> unicode

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

class ValidationError (errors: Sequence[Union[Sequence[Any], pydantic.error_wrappers.ErrorWrapper]], model: ModelOrDc)

Mixin to provide str, repr, and pretty methods. See #884 for more details.

pretty is used by devtools to provide human readable representations of objects.

Ancestors

• pydantic.utils.Representation
• builtins.ValueError
• builtins.Exception
• builtins.BaseException

Instance variables

var model

Return an attribute of instance, which is of type owner.

var raw_errors

Return an attribute of instance, which is of type owner.

Methods

def errors(self) ‑> List[ErrorDict]

def json(self, *, indent: Union[None, int, str] = 2) ‑> unicode