Module aws_lambda_powertools.utilities.data_classes.sns_event

Classes

class SNSEvent (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

SNS Event

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html

Parameters

data : Dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class SNSEvent(DictWrapper):
    """SNS Event

    Documentation:
    -------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html
    """

    @property
    def records(self) -> Iterator[SNSEventRecord]:
        for record in self["Records"]:
            yield SNSEventRecord(record)

    @property
    def record(self) -> SNSEventRecord:
        """Return the first SNS event record"""
        return next(self.records)

    @property
    def sns_message(self) -> str:
        """Return the message for the first sns event record"""
        return self.record.sns.message

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container

Instance variables

var record : SNSEventRecord

Return the first SNS event record

Expand source code

@property
def record(self) -> SNSEventRecord:
    """Return the first SNS event record"""
    return next(self.records)

var records : Iterator[SNSEventRecord]

Expand source code

@property
def records(self) -> Iterator[SNSEventRecord]:
    for record in self["Records"]:
        yield SNSEventRecord(record)

var sns_message : str

Return the message for the first sns event record

Expand source code

@property
def sns_message(self) -> str:
    """Return the message for the first sns event record"""
    return self.record.sns.message

Inherited members

• DictWrapper:
  • get
  • raw_event

class SNSEventRecord (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class SNSEventRecord(DictWrapper):
    @property
    def event_version(self) -> str:
        """Event version"""
        return self["EventVersion"]

    @property
    def event_subscription_arn(self) -> str:
        return self["EventSubscriptionArn"]

    @property
    def event_source(self) -> str:
        """The AWS service from which the SNS event record originated. For SNS, this is aws:sns"""
        return self["EventSource"]

    @property
    def sns(self) -> SNSMessage:
        return SNSMessage(self._data["Sns"])

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container

Instance variables

var event_source : str

The AWS service from which the SNS event record originated. For SNS, this is aws:sns

Expand source code

@property
def event_source(self) -> str:
    """The AWS service from which the SNS event record originated. For SNS, this is aws:sns"""
    return self["EventSource"]

var event_subscription_arn : str

Expand source code

@property
def event_subscription_arn(self) -> str:
    return self["EventSubscriptionArn"]

var event_version : str

Event version

Expand source code

@property
def event_version(self) -> str:
    """Event version"""
    return self["EventVersion"]

var sns : SNSMessage

Expand source code

@property
def sns(self) -> SNSMessage:
    return SNSMessage(self._data["Sns"])

Inherited members

• DictWrapper:
  • get
  • raw_event

class SNSMessage (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class SNSMessage(DictWrapper):
    @property
    def signature_version(self) -> str:
        """Version of the Amazon SNS signature used."""
        return self["SignatureVersion"]

    @property
    def timestamp(self) -> str:
        """The time (GMT) when the subscription confirmation was sent."""
        return self["Timestamp"]

    @property
    def signature(self) -> str:
        """Base64-encoded "SHA1withRSA" signature of the Message, MessageId, Type, Timestamp, and TopicArn values."""
        return self["Signature"]

    @property
    def signing_cert_url(self) -> str:
        """The URL to the certificate that was used to sign the message."""
        return self["SigningCertUrl"]

    @property
    def message_id(self) -> str:
        """A Universally Unique Identifier, unique for each message published.

        For a message that Amazon SNS resends during a retry, the message ID of the original message is used."""
        return self["MessageId"]

    @property
    def message(self) -> str:
        """A string that describes the message."""
        return self["Message"]

    @property
    def message_attributes(self) -> Dict[str, SNSMessageAttribute]:
        return {k: SNSMessageAttribute(v) for (k, v) in self["MessageAttributes"].items()}

    @property
    def get_type(self) -> str:
        """The type of message.

        For a subscription confirmation, the type is SubscriptionConfirmation."""
        # Note: this name conflicts with existing python builtins
        return self["Type"]

    @property
    def unsubscribe_url(self) -> str:
        """A URL that you can use to unsubscribe the endpoint from this topic.

        If you visit this URL, Amazon SNS unsubscribes the endpoint and stops sending notifications to this endpoint."""
        return self["UnsubscribeUrl"]

    @property
    def topic_arn(self) -> str:
        """The Amazon Resource Name (ARN) for the topic that this endpoint is subscribed to."""
        return self["TopicArn"]

    @property
    def subject(self) -> str:
        """The Subject parameter specified when the notification was published to the topic."""
        return self["Subject"]

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container

Instance variables

var get_type : str

The type of message.

For a subscription confirmation, the type is SubscriptionConfirmation.

Expand source code

@property
def get_type(self) -> str:
    """The type of message.

    For a subscription confirmation, the type is SubscriptionConfirmation."""
    # Note: this name conflicts with existing python builtins
    return self["Type"]

var message : str

A string that describes the message.

Expand source code

@property
def message(self) -> str:
    """A string that describes the message."""
    return self["Message"]

var message_attributes : Dict[str, SNSMessageAttribute]

Expand source code

@property
def message_attributes(self) -> Dict[str, SNSMessageAttribute]:
    return {k: SNSMessageAttribute(v) for (k, v) in self["MessageAttributes"].items()}

var message_id : str

A Universally Unique Identifier, unique for each message published.

For a message that Amazon SNS resends during a retry, the message ID of the original message is used.

Expand source code

@property
def message_id(self) -> str:
    """A Universally Unique Identifier, unique for each message published.

    For a message that Amazon SNS resends during a retry, the message ID of the original message is used."""
    return self["MessageId"]

var signature : str

Base64-encoded "SHA1withRSA" signature of the Message, MessageId, Type, Timestamp, and TopicArn values.

Expand source code

@property
def signature(self) -> str:
    """Base64-encoded "SHA1withRSA" signature of the Message, MessageId, Type, Timestamp, and TopicArn values."""
    return self["Signature"]

var signature_version : str

Version of the Amazon SNS signature used.

Expand source code

@property
def signature_version(self) -> str:
    """Version of the Amazon SNS signature used."""
    return self["SignatureVersion"]

var signing_cert_url : str

The URL to the certificate that was used to sign the message.

Expand source code

@property
def signing_cert_url(self) -> str:
    """The URL to the certificate that was used to sign the message."""
    return self["SigningCertUrl"]

var subject : str

The Subject parameter specified when the notification was published to the topic.

Expand source code

@property
def subject(self) -> str:
    """The Subject parameter specified when the notification was published to the topic."""
    return self["Subject"]

var timestamp : str

The time (GMT) when the subscription confirmation was sent.

Expand source code

@property
def timestamp(self) -> str:
    """The time (GMT) when the subscription confirmation was sent."""
    return self["Timestamp"]

var topic_arn : str

The Amazon Resource Name (ARN) for the topic that this endpoint is subscribed to.

Expand source code

@property
def topic_arn(self) -> str:
    """The Amazon Resource Name (ARN) for the topic that this endpoint is subscribed to."""
    return self["TopicArn"]

var unsubscribe_url : str

A URL that you can use to unsubscribe the endpoint from this topic.

If you visit this URL, Amazon SNS unsubscribes the endpoint and stops sending notifications to this endpoint.

Expand source code

@property
def unsubscribe_url(self) -> str:
    """A URL that you can use to unsubscribe the endpoint from this topic.

    If you visit this URL, Amazon SNS unsubscribes the endpoint and stops sending notifications to this endpoint."""
    return self["UnsubscribeUrl"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class SNSMessageAttribute (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class SNSMessageAttribute(DictWrapper):
    @property
    def get_type(self) -> str:
        """The supported message attribute data types are String, String.Array, Number, and Binary."""
        # Note: this name conflicts with existing python builtins
        return self["Type"]

    @property
    def value(self) -> str:
        """The user-specified message attribute value."""
        return self["Value"]

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container

Instance variables

var get_type : str

The supported message attribute data types are String, String.Array, Number, and Binary.

Expand source code

@property
def get_type(self) -> str:
    """The supported message attribute data types are String, String.Array, Number, and Binary."""
    # Note: this name conflicts with existing python builtins
    return self["Type"]

var value : str

The user-specified message attribute value.

Expand source code

@property
def value(self) -> str:
    """The user-specified message attribute value."""
    return self["Value"]

Inherited members

• DictWrapper:
  • get
  • raw_event