Module aws_lambda_powertools.utilities.parser

Advanced event_parser utility

Expand source code 
"""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.compat
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: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), Any, ForwardRef(None)] = None, include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), Any, ForwardRef(None)] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = 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 minLength 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, LambdaContext], ~EventParserReturnType], event: Dict[str, Any], context: LambdaContext, model: Optional[Type[~Model]] = None, 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 or is not provided
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: Optional[Type[Model]] = None,
    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 or is not provided
    InvalidEnvelopeError
        When envelope given does not implement BaseEnvelope
    """

    # The first parameter of a Lambda function is always the event
    # This line get the model informed in the event_parser function
    # or the first parameter of the function by using typing.get_type_hints
    type_hints = typing.get_type_hints(handler)
    model = model or (list(type_hints.values())[0] if type_hints else None)
    if model is None:
        raise InvalidModelTypeError(
            "The model must be provided either as the `model` argument to `event_parser`"
            "or as the type hint of `event` in the handler that it wraps",
        )

    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:
        disable_pydantic_v2_warning()
        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: Optional[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
        """
        disable_pydantic_v2_warning()

        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

Methods

def parse(self, data: Union[Dict[str, Any], Any, ForwardRef(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

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'), ForwardRef(None)] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), ForwardRef(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'), ForwardRef(None)] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), ForwardRef(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'), ForwardRef(None)] = None, exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), ForwardRef(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[ForwardRef(None), int, str] = 2) ‑> unicode