Module aws_lambda_powertools.utilities.parser.models

Expand source code 
from .alb import AlbModel, AlbRequestContext, AlbRequestContextData
from .cloudwatch import CloudWatchLogsData, CloudWatchLogsDecode, CloudWatchLogsLogEvent, CloudWatchLogsModel
from .dynamodb import DynamoDBStreamChangedRecordModel, DynamoDBStreamModel, DynamoDBStreamRecordModel
from .event_bridge import EventBridgeModel
from .kinesis import KinesisDataStreamModel, KinesisDataStreamRecord, KinesisDataStreamRecordPayload
from .s3 import S3Model, S3RecordModel
from .s3_object_event import (
    S3ObjectConfiguration,
    S3ObjectContext,
    S3ObjectLambdaEvent,
    S3ObjectSessionAttributes,
    S3ObjectSessionContext,
    S3ObjectSessionIssuer,
    S3ObjectUserIdentity,
    S3ObjectUserRequest,
)
from .ses import (
    SesMail,
    SesMailCommonHeaders,
    SesMailHeaders,
    SesMessage,
    SesModel,
    SesReceipt,
    SesReceiptAction,
    SesReceiptVerdict,
    SesRecordModel,
)
from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
from .sqs import SqsAttributesModel, SqsModel, SqsMsgAttributeModel, SqsRecordModel

__all__ = [
    "CloudWatchLogsData",
    "CloudWatchLogsDecode",
    "CloudWatchLogsLogEvent",
    "CloudWatchLogsModel",
    "AlbModel",
    "AlbRequestContext",
    "AlbRequestContextData",
    "DynamoDBStreamModel",
    "EventBridgeModel",
    "DynamoDBStreamChangedRecordModel",
    "DynamoDBStreamRecordModel",
    "DynamoDBStreamChangedRecordModel",
    "KinesisDataStreamModel",
    "KinesisDataStreamRecord",
    "KinesisDataStreamRecordPayload",
    "S3Model",
    "S3RecordModel",
    "S3ObjectLambdaEvent",
    "S3ObjectUserIdentity",
    "S3ObjectSessionContext",
    "S3ObjectSessionAttributes",
    "S3ObjectSessionIssuer",
    "S3ObjectUserRequest",
    "S3ObjectConfiguration",
    "S3ObjectContext",
    "SesModel",
    "SesRecordModel",
    "SesMessage",
    "SesMail",
    "SesMailCommonHeaders",
    "SesMailHeaders",
    "SesReceipt",
    "SesReceiptAction",
    "SesReceiptVerdict",
    "SnsModel",
    "SnsNotificationModel",
    "SnsRecordModel",
    "SqsModel",
    "SqsRecordModel",
    "SqsMsgAttributeModel",
    "SqsAttributesModel",
]

Sub-modules

aws_lambda_powertools.utilities.parser.models.alb
aws_lambda_powertools.utilities.parser.models.cloudwatch
aws_lambda_powertools.utilities.parser.models.dynamodb
aws_lambda_powertools.utilities.parser.models.event_bridge
aws_lambda_powertools.utilities.parser.models.kinesis
aws_lambda_powertools.utilities.parser.models.s3
aws_lambda_powertools.utilities.parser.models.s3_object_event
aws_lambda_powertools.utilities.parser.models.ses
aws_lambda_powertools.utilities.parser.models.sns
aws_lambda_powertools.utilities.parser.models.sqs

Classes

class AlbModel (**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 AlbModel(BaseModel):
    httpMethod: str
    path: str
    body: str
    isBase64Encoded: bool
    headers: Dict[str, str]
    queryStringParameters: Dict[str, str]
    requestContext: AlbRequestContext

Ancestors

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

Class variables

var body : str
var headers : Dict[str, str]
var httpMethod : str
var isBase64Encoded : bool
var path : str
var queryStringParameters : Dict[str, str]
var requestContext : AlbRequestContext
class AlbRequestContext (**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 AlbRequestContext(BaseModel):
    elb: AlbRequestContextData

Ancestors

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

Class variables

var elb : AlbRequestContextData
class AlbRequestContextData (**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 AlbRequestContextData(BaseModel):
    targetGroupArn: str

Ancestors

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

Class variables

var targetGroupArn : str
class CloudWatchLogsData (**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 CloudWatchLogsData(BaseModel):
    decoded_data: CloudWatchLogsDecode = Field(None, alias="data")

    @validator("decoded_data", pre=True)
    def prepare_data(cls, value):
        try:
            logger.debug("Decoding base64 cloudwatch log data before parsing")
            payload = base64.b64decode(value)
            logger.debug("Decompressing cloudwatch log data before parsing")
            uncompressed = zlib.decompress(payload, zlib.MAX_WBITS | 32)
            return json.loads(uncompressed.decode("utf-8"))
        except Exception:
            raise ValueError("unable to decompress data")

Ancestors

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

Class variables

var decoded_data : CloudWatchLogsDecode

Static methods

def prepare_data(value)
Expand source code 
@validator("decoded_data", pre=True)
def prepare_data(cls, value):
    try:
        logger.debug("Decoding base64 cloudwatch log data before parsing")
        payload = base64.b64decode(value)
        logger.debug("Decompressing cloudwatch log data before parsing")
        uncompressed = zlib.decompress(payload, zlib.MAX_WBITS | 32)
        return json.loads(uncompressed.decode("utf-8"))
    except Exception:
        raise ValueError("unable to decompress data")
class CloudWatchLogsDecode (**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 CloudWatchLogsDecode(BaseModel):
    messageType: str
    owner: str
    logGroup: str
    logStream: str
    subscriptionFilters: List[str]
    logEvents: List[CloudWatchLogsLogEvent]

Ancestors

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

Class variables

var logEvents : List[CloudWatchLogsLogEvent]
var logGroup : str
var logStream : str
var messageType : str
var owner : str
var subscriptionFilters : List[str]
class CloudWatchLogsLogEvent (**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 CloudWatchLogsLogEvent(BaseModel):
    id: str  # noqa AA03 VNE003
    timestamp: datetime
    message: str

Ancestors

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

Class variables

var id : str
var message : str
var timestamp : datetime.datetime
class CloudWatchLogsModel (**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 CloudWatchLogsModel(BaseModel):
    awslogs: CloudWatchLogsData

Ancestors

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

Class variables

var awslogs : CloudWatchLogsData
class DynamoDBStreamChangedRecordModel (**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 DynamoDBStreamChangedRecordModel(BaseModel):
    ApproximateCreationDateTime: Optional[date]
    Keys: Dict[str, Dict[str, Any]]
    NewImage: Optional[Dict[str, Any]]
    OldImage: Optional[Dict[str, Any]]
    SequenceNumber: str
    SizeBytes: int
    StreamViewType: Literal["NEW_AND_OLD_IMAGES", "KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE"]

    # context on why it's commented: https://github.com/awslabs/aws-lambda-powertools-python/pull/118
    # since both images are optional, they can both be None. However, at least one must
    # exist in a legal model of NEW_AND_OLD_IMAGES type
    # @root_validator
    # def check_one_image_exists(cls, values): # noqa: E800
    #     new_img, old_img = values.get("NewImage"), values.get("OldImage") # noqa: E800
    #     stream_type = values.get("StreamViewType") # noqa: E800
    #     if stream_type == "NEW_AND_OLD_IMAGES" and not new_img and not old_img: # noqa: E800
    #         raise TypeError("DynamoDB streams model failed validation, missing both new & old stream images") # noqa: E800,E501
    #     return values # noqa: E800

Ancestors

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

Class variables

var ApproximateCreationDateTime : Union[datetime.date, NoneType]
var Keys : Dict[str, Dict[str, Any]]
var NewImage : Union[Dict[str, Any], NoneType]
var OldImage : Union[Dict[str, Any], NoneType]
var SequenceNumber : str
var SizeBytes : int
var StreamViewType : Literal['NEW_AND_OLD_IMAGES', 'KEYS_ONLY', 'NEW_IMAGE', 'OLD_IMAGE']
class DynamoDBStreamModel (**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 DynamoDBStreamModel(BaseModel):
    Records: List[DynamoDBStreamRecordModel]

Ancestors

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

Class variables

var Records : List[DynamoDBStreamRecordModel]
class DynamoDBStreamRecordModel (**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 DynamoDBStreamRecordModel(BaseModel):
    eventID: str
    eventName: Literal["INSERT", "MODIFY", "REMOVE"]
    eventVersion: float
    eventSource: Literal["aws:dynamodb"]
    awsRegion: str
    eventSourceARN: str
    dynamodb: DynamoDBStreamChangedRecordModel
    userIdentity: Optional[UserIdentity]

Ancestors

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

Class variables

var awsRegion : str
var dynamodb : DynamoDBStreamChangedRecordModel
var eventID : str
var eventName : Literal['INSERT', 'MODIFY', 'REMOVE']
var eventSource : Literal['aws:aws_lambda_powertools.utilities.parser.models.dynamodb']
var eventSourceARN : str
var eventVersion : float
var userIdentity : Union[UserIdentity, NoneType]
class EventBridgeModel (**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 EventBridgeModel(BaseModel):
    version: str
    id: str  # noqa: A003,VNE003
    source: str
    account: str
    time: datetime
    region: str
    resources: List[str]
    detail_type: str = Field(None, alias="detail-type")
    detail: Dict[str, Any]
    replay_name: Optional[str] = Field(None, alias="replay-name")

Ancestors

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

Class variables

var account : str
var detail : Dict[str, Any]
var detail_type : str
var id : str
var region : str
var replay_name : Union[str, NoneType]
var resources : List[str]
var source : str
var time : datetime.datetime
var version : str
class KinesisDataStreamModel (**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 KinesisDataStreamModel(BaseModel):
    Records: List[KinesisDataStreamRecord]

Ancestors

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

Class variables

var Records : List[KinesisDataStreamRecord]
class KinesisDataStreamRecord (**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 KinesisDataStreamRecord(BaseModel):
    eventSource: Literal["aws:kinesis"]
    eventVersion: str
    eventID: str
    eventName: Literal["aws:kinesis:record"]
    invokeIdentityArn: str
    awsRegion: str
    eventSourceARN: str
    kinesis: KinesisDataStreamRecordPayload

Ancestors

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

Class variables

var awsRegion : str
var eventID : str
var eventName : Literal['aws:aws_lambda_powertools.utilities.parser.models.kinesis:record']
var eventSource : Literal['aws:aws_lambda_powertools.utilities.parser.models.kinesis']
var eventSourceARN : str
var eventVersion : str
var invokeIdentityArn : str
var kinesis : KinesisDataStreamRecordPayload
class KinesisDataStreamRecordPayload (**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 KinesisDataStreamRecordPayload(BaseModel):
    kinesisSchemaVersion: str
    partitionKey: str
    sequenceNumber: PositiveInt
    data: bytes  # base64 encoded str is parsed into bytes
    approximateArrivalTimestamp: float

    @validator("data", pre=True)
    def data_base64_decode(cls, value):
        try:
            logger.debug("Decoding base64 Kinesis data record before parsing")
            return base64.b64decode(value)
        except (BinAsciiError, TypeError):
            raise ValueError("base64 decode failed")

Ancestors

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

Class variables

var approximateArrivalTimestamp : float
var data : bytes
var kinesisSchemaVersion : str
var partitionKey : str
var sequenceNumber : pydantic.types.PositiveInt

Static methods

def data_base64_decode(value)
Expand source code 
@validator("data", pre=True)
def data_base64_decode(cls, value):
    try:
        logger.debug("Decoding base64 Kinesis data record before parsing")
        return base64.b64decode(value)
    except (BinAsciiError, TypeError):
        raise ValueError("base64 decode failed")
class S3Model (**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 S3Model(BaseModel):
    Records: List[S3RecordModel]

Ancestors

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

Class variables

var Records : List[S3RecordModel]
class S3ObjectConfiguration (**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 S3ObjectConfiguration(BaseModel):
    accessPointArn: str
    supportingAccessPointArn: str
    payload: str

Ancestors

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

Class variables

var accessPointArn : str
var payload : str
var supportingAccessPointArn : str
class S3ObjectContext (**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 S3ObjectContext(BaseModel):
    inputS3Url: HttpUrl
    outputRoute: str
    outputToken: str

Ancestors

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

Class variables

var inputS3Url : pydantic.networks.HttpUrl
var outputRoute : str
var outputToken : str
class S3ObjectLambdaEvent (**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 S3ObjectLambdaEvent(BaseModel):
    xAmzRequestId: str
    getObjectContext: S3ObjectContext
    configuration: S3ObjectConfiguration
    userRequest: S3ObjectUserRequest
    userIdentity: S3ObjectUserIdentity
    protocolVersion: str

Ancestors

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

Class variables

var configuration : S3ObjectConfiguration
var getObjectContext : S3ObjectContext
var protocolVersion : str
var userIdentity : S3ObjectUserIdentity
var userRequest : S3ObjectUserRequest
var xAmzRequestId : str
class S3ObjectSessionAttributes (**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 S3ObjectSessionAttributes(BaseModel):
    creationDate: str
    mfaAuthenticated: bool

Ancestors

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

Class variables

var creationDate : str
var mfaAuthenticated : bool
class S3ObjectSessionContext (**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 S3ObjectSessionContext(BaseModel):
    sessionIssuer: S3ObjectSessionIssuer
    attributes: S3ObjectSessionAttributes

Ancestors

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

Class variables

var attributes : S3ObjectSessionAttributes
var sessionIssuer : S3ObjectSessionIssuer
class S3ObjectSessionIssuer (**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 S3ObjectSessionIssuer(BaseModel):
    type: str  # noqa: A003, VNE003
    userName: Optional[str]
    principalId: str
    arn: str
    accountId: str

Ancestors

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

Class variables

var accountId : str
var arn : str
var principalId : str
var type : str
var userName : Union[str, NoneType]
class S3ObjectUserIdentity (**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 S3ObjectUserIdentity(BaseModel):
    type: str  # noqa003
    accountId: str
    accessKeyId: str
    userName: Optional[str]
    principalId: str
    arn: str
    sessionContext: Optional[S3ObjectSessionContext]

Ancestors

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

Class variables

var accessKeyId : str
var accountId : str
var arn : str
var principalId : str
var sessionContext : Union[S3ObjectSessionContext, NoneType]
var type : str
var userName : Union[str, NoneType]
class S3ObjectUserRequest (**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 S3ObjectUserRequest(BaseModel):
    url: str
    headers: Dict[str, str]

Ancestors

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

Class variables

var headers : Dict[str, str]
var url : str
class S3RecordModel (**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 S3RecordModel(BaseModel):
    eventVersion: str
    eventSource: Literal["aws:s3"]
    awsRegion: str
    eventTime: datetime
    eventName: str
    userIdentity: S3Identity
    requestParameters: S3RequestParameters
    responseElements: S3ResponseElements
    s3: S3Message
    glacierEventData: Optional[S3EventRecordGlacierEventData]

Ancestors

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

Class variables

var awsRegion : str
var eventName : str
var eventSource : Literal['aws:aws_lambda_powertools.utilities.parser.models.s3']
var eventTime : datetime.datetime
var eventVersion : str
var glacierEventData : Union[S3EventRecordGlacierEventData, NoneType]
var requestParameters : S3RequestParameters
var responseElements : S3ResponseElements
var s3 : S3Message
var userIdentity : S3Identity
class SesMail (**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 SesMail(BaseModel):
    timestamp: datetime
    source: EmailStr
    messageId: str
    destination: List[EmailStr]
    headersTruncated: bool
    headers: List[SesMailHeaders]
    commonHeaders: SesMailCommonHeaders

Ancestors

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

Class variables

var commonHeaders : SesMailCommonHeaders
var destination : List[pydantic.networks.EmailStr]
var headers : List[SesMailHeaders]
var headersTruncated : bool
var messageId : str
var source : pydantic.networks.EmailStr
var timestamp : datetime.datetime
class SesMailCommonHeaders (**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 SesMailCommonHeaders(BaseModel):
    header_from: List[str] = Field(None, alias="from")
    to: List[str]
    cc: Optional[List[str]]
    bcc: Optional[List[str]]
    sender: Optional[List[str]]
    reply_to: Optional[List[str]] = Field(None, alias="reply-to")
    returnPath: EmailStr
    messageId: str
    date: str
    subject: str

Ancestors

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

Class variables

var bcc : Union[List[str], NoneType]
var cc : Union[List[str], NoneType]
var date : str
var header_from : List[str]
var messageId : str
var reply_to : Union[List[str], NoneType]
var returnPath : pydantic.networks.EmailStr
var sender : Union[List[str], NoneType]
var subject : str
var to : List[str]
class SesMailHeaders (**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 SesMailHeaders(BaseModel):
    name: str
    value: str

Ancestors

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

Class variables

var name : str
var value : str
class SesMessage (**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 SesMessage(BaseModel):
    mail: SesMail
    receipt: SesReceipt

Ancestors

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

Class variables

var mail : SesMail
var receipt : SesReceipt
class SesModel (**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 SesModel(BaseModel):
    Records: List[SesRecordModel]

Ancestors

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

Class variables

var Records : List[SesRecordModel]
class SesReceipt (**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 SesReceipt(BaseModel):
    timestamp: datetime
    processingTimeMillis: PositiveInt
    recipients: List[EmailStr]
    spamVerdict: SesReceiptVerdict
    virusVerdict: SesReceiptVerdict
    spfVerdict: SesReceiptVerdict
    dmarcVerdict: SesReceiptVerdict
    action: SesReceiptAction

Ancestors

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

Class variables

var action : SesReceiptAction
var dmarcVerdict : SesReceiptVerdict
var processingTimeMillis : pydantic.types.PositiveInt
var recipients : List[pydantic.networks.EmailStr]
var spamVerdict : SesReceiptVerdict
var spfVerdict : SesReceiptVerdict
var timestamp : datetime.datetime
var virusVerdict : SesReceiptVerdict
class SesReceiptAction (**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 SesReceiptAction(BaseModel):
    type: Literal["Lambda"]  # noqa A003,VNE003
    invocationType: Literal["Event"]
    functionArn: str

Ancestors

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

Class variables

var functionArn : str
var invocationType : Literal['Event']
var type : Literal['Lambda']
class SesReceiptVerdict (**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 SesReceiptVerdict(BaseModel):
    status: Literal["PASS", "FAIL", "GRAY", "PROCESSING_FAILED"]

Ancestors

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

Class variables

var status : Literal['PASS', 'FAIL', 'GRAY', 'PROCESSING_FAILED']
class SesRecordModel (**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 SesRecordModel(BaseModel):
    eventSource: Literal["aws:ses"]
    eventVersion: str
    ses: SesMessage

Ancestors

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

Class variables

var eventSource : Literal['aws:aws_lambda_powertools.utilities.parser.models.ses']
var eventVersion : str
var ses : SesMessage
class SnsModel (**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 SnsModel(BaseModel):
    Records: List[SnsRecordModel]

Ancestors

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

Class variables

var Records : List[SnsRecordModel]
class SnsNotificationModel (**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 SnsNotificationModel(BaseModel):
    Subject: Optional[str]
    TopicArn: str
    UnsubscribeUrl: HttpUrl
    Type: Literal["Notification"]
    MessageAttributes: Optional[Dict[str, SnsMsgAttributeModel]]
    Message: str
    MessageId: str
    SigningCertUrl: HttpUrl
    Signature: str
    Timestamp: datetime
    SignatureVersion: str

    @root_validator(pre=True)
    def check_sqs_protocol(cls, values):
        sqs_rewritten_keys = ("UnsubscribeURL", "SigningCertURL")
        if any(key in sqs_rewritten_keys for key in values):
            values["UnsubscribeUrl"] = values.pop("UnsubscribeURL")
            values["SigningCertUrl"] = values.pop("SigningCertURL")
        return values

Ancestors

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

Class variables

var Message : str
var MessageAttributes : Union[Dict[str, SnsMsgAttributeModel], NoneType]
var MessageId : str
var Signature : str
var SignatureVersion : str
var SigningCertUrl : pydantic.networks.HttpUrl
var Subject : Union[str, NoneType]
var Timestamp : datetime.datetime
var TopicArn : str
var Type : Literal['Notification']
var UnsubscribeUrl : pydantic.networks.HttpUrl

Static methods

def check_sqs_protocol(values)
Expand source code 
@root_validator(pre=True)
def check_sqs_protocol(cls, values):
    sqs_rewritten_keys = ("UnsubscribeURL", "SigningCertURL")
    if any(key in sqs_rewritten_keys for key in values):
        values["UnsubscribeUrl"] = values.pop("UnsubscribeURL")
        values["SigningCertUrl"] = values.pop("SigningCertURL")
    return values
class SnsRecordModel (**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 SnsRecordModel(BaseModel):
    EventSource: Literal["aws:sns"]
    EventVersion: str
    EventSubscriptionArn: str
    Sns: SnsNotificationModel

Ancestors

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

Class variables

var EventSource : Literal['aws:aws_lambda_powertools.utilities.parser.models.sns']
var EventSubscriptionArn : str
var EventVersion : str
var Sns : SnsNotificationModel
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 attributes : SqsAttributesModel
var awsRegion : str
var body : str
var eventSource : Literal['aws:aws_lambda_powertools.utilities.parser.models.sqs']
var eventSourceARN : str
var md5OfBody : str
var md5OfMessageAttributes : Union[str, NoneType]
var messageAttributes : Dict[str, SqsMsgAttributeModel]
var messageId : str
var receiptHandle : str