Module aws_lambda_powertools.utilities.parser.models.sqs

Expand source code
from datetime import datetime
from typing import Dict, List, Optional

from pydantic import BaseModel

from ..types import Literal


class SqsAttributesModel(BaseModel):
    ApproximateReceiveCount: str
    ApproximateFirstReceiveTimestamp: datetime
    MessageDeduplicationId: Optional[str]
    MessageGroupId: Optional[str]
    SenderId: str
    SentTimestamp: datetime
    SequenceNumber: Optional[str]
    AWSTraceHeader: Optional[str]


class SqsMsgAttributeModel(BaseModel):
    stringValue: Optional[str]
    binaryValue: Optional[str]
    stringListValues: List[str] = []
    binaryListValues: List[str] = []
    dataType: str

    # context on why it's commented: https://github.com/awslabs/aws-lambda-powertools-python/pull/118
    # Amazon SQS supports the logical data types String, Number, and Binary with optional custom data type
    # labels with the format .custom-data-type.
    # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html#sqs-message-attributes
    # @validator("dataType")
    # def valid_type(cls, v):  # noqa: VNE001,E800 # noqa: E800
    #     pattern = re.compile("Number.*|String.*|Binary.*") # noqa: E800
    #     if not pattern.match(v): # noqa: E800
    #         raise TypeError("data type is invalid") # noqa: E800
    #     return v # noqa: E800
    #
    # # validate that dataType and value are not None and match
    # @root_validator
    # def check_str_and_binary_values(cls, values): # noqa: E800
    #     binary_val, str_val = values.get("binaryValue", ""), values.get("stringValue", "") # noqa: E800
    #     data_type = values.get("dataType") # noqa: E800
    #     if not str_val and not binary_val: # noqa: E800
    #         raise TypeError("both binaryValue and stringValue are missing") # noqa: E800
    #     if data_type.startswith("Binary") and not binary_val: # noqa: E800
    #         raise TypeError("binaryValue is missing") # noqa: E800
    #     if (data_type.startswith("String") or data_type.startswith("Number")) and not str_val: # noqa: E800
    #         raise TypeError("stringValue is missing") # noqa: E800
    #     return values # noqa: E800


class SqsRecordModel(BaseModel):
    messageId: str
    receiptHandle: str
    body: str
    attributes: SqsAttributesModel
    messageAttributes: Dict[str, SqsMsgAttributeModel]
    md5OfBody: str
    md5OfMessageAttributes: Optional[str]
    eventSource: Literal["aws:sqs"]
    eventSourceARN: str
    awsRegion: str


class SqsModel(BaseModel):
    Records: List[SqsRecordModel]

Classes

class SqsAttributesModel (**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.

Expand source code
class SqsAttributesModel(BaseModel):
    ApproximateReceiveCount: str
    ApproximateFirstReceiveTimestamp: datetime
    MessageDeduplicationId: Optional[str]
    MessageGroupId: Optional[str]
    SenderId: str
    SentTimestamp: datetime
    SequenceNumber: Optional[str]
    AWSTraceHeader: Optional[str]

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var AWSTraceHeader : Union[str, NoneType]
var ApproximateFirstReceiveTimestamp : datetime.datetime
var ApproximateReceiveCount : str
var MessageDeduplicationId : Union[str, NoneType]
var MessageGroupId : Union[str, NoneType]
var SenderId : str
var SentTimestamp : datetime.datetime
var SequenceNumber : Union[str, NoneType]
class SqsModel (**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.

Expand source code
class SqsModel(BaseModel):
    Records: List[SqsRecordModel]

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var Records : List[SqsRecordModel]
class SqsMsgAttributeModel (**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.

Expand source code
class SqsMsgAttributeModel(BaseModel):
    stringValue: Optional[str]
    binaryValue: Optional[str]
    stringListValues: List[str] = []
    binaryListValues: List[str] = []
    dataType: str

    # context on why it's commented: https://github.com/awslabs/aws-lambda-powertools-python/pull/118
    # Amazon SQS supports the logical data types String, Number, and Binary with optional custom data type
    # labels with the format .custom-data-type.
    # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html#sqs-message-attributes
    # @validator("dataType")
    # def valid_type(cls, v):  # noqa: VNE001,E800 # noqa: E800
    #     pattern = re.compile("Number.*|String.*|Binary.*") # noqa: E800
    #     if not pattern.match(v): # noqa: E800
    #         raise TypeError("data type is invalid") # noqa: E800
    #     return v # noqa: E800
    #
    # # validate that dataType and value are not None and match
    # @root_validator
    # def check_str_and_binary_values(cls, values): # noqa: E800
    #     binary_val, str_val = values.get("binaryValue", ""), values.get("stringValue", "") # noqa: E800
    #     data_type = values.get("dataType") # noqa: E800
    #     if not str_val and not binary_val: # noqa: E800
    #         raise TypeError("both binaryValue and stringValue are missing") # noqa: E800
    #     if data_type.startswith("Binary") and not binary_val: # noqa: E800
    #         raise TypeError("binaryValue is missing") # noqa: E800
    #     if (data_type.startswith("String") or data_type.startswith("Number")) and not str_val: # noqa: E800
    #         raise TypeError("stringValue is missing") # noqa: E800
    #     return values # noqa: E800

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var binaryListValues : List[str]
var binaryValue : Union[str, NoneType]
var dataType : str
var stringListValues : List[str]
var stringValue : Union[str, NoneType]
class SqsRecordModel (**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.

Expand source code
class SqsRecordModel(BaseModel):
    messageId: str
    receiptHandle: str
    body: str
    attributes: SqsAttributesModel
    messageAttributes: Dict[str, SqsMsgAttributeModel]
    md5OfBody: str
    md5OfMessageAttributes: Optional[str]
    eventSource: Literal["aws:sqs"]
    eventSourceARN: str
    awsRegion: str

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var attributesSqsAttributesModel
var awsRegion : str
var body : str
var eventSource : Literal['aws:sqs']
var eventSourceARN : str
var md5OfBody : str
var md5OfMessageAttributes : Union[str, NoneType]
var messageAttributes : Dict[str, SqsMsgAttributeModel]
var messageId : str
var receiptHandle : str