Module aws_lambda_powertools.utilities.parser.models

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(BaseModel):
    elb: AlbRequestContextData

Ancestors

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

Class variables

var elb : AlbRequestContextData

class AlbRequestContextData(BaseModel):
    targetGroupArn: str

Ancestors

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

Class variables

var targetGroupArn : str

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(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(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(BaseModel):
    awslogs: CloudWatchLogsData

Ancestors

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

Class variables

var awslogs : CloudWatchLogsData

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(BaseModel):
    Records: List[DynamoDBStreamRecordModel]

Ancestors

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

Class variables

var Records : List[DynamoDBStreamRecordModel]

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(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(BaseModel):
    Records: List[KinesisDataStreamRecord]

Ancestors

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

Class variables

var Records : List[KinesisDataStreamRecord]

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(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(BaseModel):
    Records: List[S3RecordModel]

Ancestors

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

Class variables

var Records : List[S3RecordModel]

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 SesModel(BaseModel):
    Records: List[SesRecordModel]

Ancestors

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

Class variables

var Records : List[SesRecordModel]

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(BaseModel):
    Records: List[SnsRecordModel]

Ancestors

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

Class variables

var Records : List[SnsRecordModel]

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(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 SqsModel(BaseModel):
    Records: List[SqsRecordModel]

Ancestors

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

Class variables

var Records : List[SqsRecordModel]

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