Module aws_lambda_powertools.utilities.parser.envelopes.base

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: dict[str, Any] | Any | None, model: type[T]) -> T | None:
        """Parses envelope data against model provided

        Parameters
        ----------
        data : dict
            Data to be parsed and validated
        model : type[T]
            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

        adapter = _retrieve_or_set_model_from_cache(model=model)

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

        return adapter.validate_python(data)

    @abstractmethod
    def parse(self, data: dict[str, Any] | Any | None, model: type[T]):
        """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
• BedrockAgentEnvelope
• CloudWatchLogsEnvelope
• DynamoDBStreamEnvelope
• EventBridgeEnvelope
• KafkaEnvelope
• KinesisDataStreamEnvelope
• KinesisFirehoseEnvelope
• LambdaFunctionUrlEnvelope
• SnsEnvelope
• SnsSqsEnvelope
• SqsEnvelope
• VpcLatticeEnvelope
• VpcLatticeV2Envelope

Methods

def parse(self, data: dict[str, Any] | Any | None, model: type[T])

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)