Module aws_lambda_powertools.utilities.parser

Advanced event_parser utility

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)

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[..., ~EventParserReturnType], event: Dict[str, Any], context: LambdaContext, model: Optional[Type[~Model]] = None, envelope: Optional[Type[~Envelope]] = None, **kwargs: Any) ‑> ~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
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
def root_validator(*, pre: bool = False, allow_reuse: bool = False, skip_on_failure: bool = False)

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)

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

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)
def parse_file(path: Union[str, pathlib.Path], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: pydantic.parse.Protocol = None, allow_pickle: bool = False)
def parse_obj(obj: Any)
def parse_raw(b: Union[str, bytes], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: pydantic.parse.Protocol = None, allow_pickle: bool = False)
def schema(by_alias: bool = True, ref_template: unicode = '#/definitions/{model}')
def schema_json(*, by_alias: bool = True, ref_template: unicode = '#/definitions/{model}', **dumps_kwargs: Any)
def update_forward_refs(**localns: Any) ‑> None

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

def validate(value: Any)

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)

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)

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)

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
var raw_errors

Methods

def errors(self)
def json(self, *, indent: Union[ForwardRef(None), int, str] = 2)