Module aws_lambda_powertools.utilities.data_classes.ses_event
Expand source code
from typing import Iterator, List
from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
class SESMailHeader(DictWrapper):
@property
def name(self) -> str:
return self["name"]
@property
def value(self) -> str:
return self["value"]
class SESMailCommonHeaders(DictWrapper):
@property
def return_path(self) -> str:
"""The values in the Return-Path header of the email."""
return self["returnPath"]
@property
def get_from(self) -> List[str]:
"""The values in the From header of the email."""
# Note: this name conflicts with existing python builtins
return self["from"]
@property
def date(self) -> List[str]:
"""The date and time when Amazon SES received the message."""
return self["date"]
@property
def to(self) -> List[str]:
"""The values in the To header of the email."""
return self["to"]
@property
def message_id(self) -> str:
"""The ID of the original message."""
return str(self["messageId"])
@property
def subject(self) -> str:
"""The value of the Subject header for the email."""
return str(self["subject"])
class SESMail(DictWrapper):
@property
def timestamp(self) -> str:
"""String that contains the time at which the email was received, in ISO8601 format."""
return self["timestamp"]
@property
def source(self) -> str:
"""String that contains the email address (specifically, the envelope MAIL FROM address)
that the email was sent from."""
return self["source"]
@property
def message_id(self) -> str:
"""String that contains the unique ID assigned to the email by Amazon SES.
If the email was delivered to Amazon S3, the message ID is also the Amazon S3 object key that was
used to write the message to your Amazon S3 bucket."""
return self["messageId"]
@property
def destination(self) -> List[str]:
"""A complete list of all recipient addresses (including To: and CC: recipients)
from the MIME headers of the incoming email."""
return self["destination"]
@property
def headers_truncated(self) -> bool:
"""String that specifies whether the headers were truncated in the notification, which will happen
if the headers are larger than 10 KB. Possible values are true and false."""
return bool(self["headersTruncated"])
@property
def headers(self) -> Iterator[SESMailHeader]:
"""A list of Amazon SES headers and your custom headers.
Each header in the list has a name field and a value field"""
for header in self["headers"]:
yield SESMailHeader(header)
@property
def common_headers(self) -> SESMailCommonHeaders:
"""A list of headers common to all emails. Each header in the list is composed of a name and a value."""
return SESMailCommonHeaders(self["commonHeaders"])
class SESReceiptStatus(DictWrapper):
@property
def status(self) -> str:
return str(self["status"])
class SESReceiptAction(DictWrapper):
@property
def get_type(self) -> str:
"""String that indicates the type of action that was executed.
Possible values are S3, SNS, Bounce, Lambda, Stop, and WorkMail
"""
# Note: this name conflicts with existing python builtins
return self["type"]
@property
def function_arn(self) -> str:
"""String that contains the ARN of the Lambda function that was triggered.
Present only for the Lambda action type."""
return self["functionArn"]
@property
def invocation_type(self) -> str:
"""String that contains the invocation type of the Lambda function. Possible values are RequestResponse
and Event. Present only for the Lambda action type."""
return self["invocationType"]
class SESReceipt(DictWrapper):
@property
def timestamp(self) -> str:
"""String that specifies the date and time at which the action was triggered, in ISO 8601 format."""
return self["timestamp"]
@property
def processing_time_millis(self) -> int:
"""String that specifies the period, in milliseconds, from the time Amazon SES received the message
to the time it triggered the action."""
return int(self["processingTimeMillis"])
@property
def recipients(self) -> List[str]:
"""A list of recipients (specifically, the envelope RCPT TO addresses) that were matched by the
active receipt rule. The addresses listed here may differ from those listed by the destination
field in the mail object."""
return self["recipients"]
@property
def spam_verdict(self) -> SESReceiptStatus:
"""Object that indicates whether the message is spam."""
return SESReceiptStatus(self["spamVerdict"])
@property
def virus_verdict(self) -> SESReceiptStatus:
"""Object that indicates whether the message contains a virus."""
return SESReceiptStatus(self["virusVerdict"])
@property
def spf_verdict(self) -> SESReceiptStatus:
"""Object that indicates whether the Sender Policy Framework (SPF) check passed."""
return SESReceiptStatus(self["spfVerdict"])
@property
def dmarc_verdict(self) -> SESReceiptStatus:
"""Object that indicates whether the Domain-based Message Authentication,
Reporting & Conformance (DMARC) check passed."""
return SESReceiptStatus(self["dmarcVerdict"])
@property
def action(self) -> SESReceiptAction:
"""Object that encapsulates information about the action that was executed."""
return SESReceiptAction(self["action"])
class SESMessage(DictWrapper):
@property
def mail(self) -> SESMail:
return SESMail(self["ses"]["mail"])
@property
def receipt(self) -> SESReceipt:
return SESReceipt(self["ses"]["receipt"])
class SESEventRecord(DictWrapper):
@property
def event_source(self) -> str:
"""The AWS service from which the SES event record originated. For SES, this is aws:ses"""
return self["eventSource"]
@property
def event_version(self) -> str:
"""The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
return self["eventVersion"]
@property
def ses(self) -> SESMessage:
return SESMessage(self._data)
class SESEvent(DictWrapper):
"""Amazon SES to receive message event trigger
NOTE: There is a 30-second timeout on RequestResponse invocations.
Documentation:
--------------
- https://docs.aws.amazon.com/lambda/latest/dg/services-ses.html
- https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-action-lambda.html
"""
@property
def records(self) -> Iterator[SESEventRecord]:
for record in self["Records"]:
yield SESEventRecord(record)
@property
def record(self) -> SESEventRecord:
return next(self.records)
@property
def mail(self) -> SESMail:
return self.record.ses.mail
@property
def receipt(self) -> SESReceipt:
return self.record.ses.receipt
Classes
class SESEvent (data: Dict[str, Any])
-
Amazon SES to receive message event trigger
NOTE: There is a 30-second timeout on RequestResponse invocations.
Documentation:
Expand source code
class SESEvent(DictWrapper): """Amazon SES to receive message event trigger NOTE: There is a 30-second timeout on RequestResponse invocations. Documentation: -------------- - https://docs.aws.amazon.com/lambda/latest/dg/services-ses.html - https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-action-lambda.html """ @property def records(self) -> Iterator[SESEventRecord]: for record in self["Records"]: yield SESEventRecord(record) @property def record(self) -> SESEventRecord: return next(self.records) @property def mail(self) -> SESMail: return self.record.ses.mail @property def receipt(self) -> SESReceipt: return self.record.ses.receipt
Ancestors
Instance variables
var mail : SESMail
-
Expand source code
@property def mail(self) -> SESMail: return self.record.ses.mail
var receipt : SESReceipt
-
Expand source code
@property def receipt(self) -> SESReceipt: return self.record.ses.receipt
var record : SESEventRecord
-
Expand source code
@property def record(self) -> SESEventRecord: return next(self.records)
var records : Iterator[SESEventRecord]
-
Expand source code
@property def records(self) -> Iterator[SESEventRecord]: for record in self["Records"]: yield SESEventRecord(record)
Inherited members
class SESEventRecord (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESEventRecord(DictWrapper): @property def event_source(self) -> str: """The AWS service from which the SES event record originated. For SES, this is aws:ses""" return self["eventSource"] @property def event_version(self) -> str: """The eventVersion key value contains a major and minor version in the form <major>.<minor>.""" return self["eventVersion"] @property def ses(self) -> SESMessage: return SESMessage(self._data)
Ancestors
Instance variables
var event_source : str
-
The AWS service from which the SES event record originated. For SES, this is aws:ses
Expand source code
@property def event_source(self) -> str: """The AWS service from which the SES event record originated. For SES, this is aws:ses""" return self["eventSource"]
var event_version : str
-
The eventVersion key value contains a major and minor version in the form
. . Expand source code
@property def event_version(self) -> str: """The eventVersion key value contains a major and minor version in the form <major>.<minor>.""" return self["eventVersion"]
var ses : SESMessage
-
Expand source code
@property def ses(self) -> SESMessage: return SESMessage(self._data)
Inherited members
class SESMail (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESMail(DictWrapper): @property def timestamp(self) -> str: """String that contains the time at which the email was received, in ISO8601 format.""" return self["timestamp"] @property def source(self) -> str: """String that contains the email address (specifically, the envelope MAIL FROM address) that the email was sent from.""" return self["source"] @property def message_id(self) -> str: """String that contains the unique ID assigned to the email by Amazon SES. If the email was delivered to Amazon S3, the message ID is also the Amazon S3 object key that was used to write the message to your Amazon S3 bucket.""" return self["messageId"] @property def destination(self) -> List[str]: """A complete list of all recipient addresses (including To: and CC: recipients) from the MIME headers of the incoming email.""" return self["destination"] @property def headers_truncated(self) -> bool: """String that specifies whether the headers were truncated in the notification, which will happen if the headers are larger than 10 KB. Possible values are true and false.""" return bool(self["headersTruncated"]) @property def headers(self) -> Iterator[SESMailHeader]: """A list of Amazon SES headers and your custom headers. Each header in the list has a name field and a value field""" for header in self["headers"]: yield SESMailHeader(header) @property def common_headers(self) -> SESMailCommonHeaders: """A list of headers common to all emails. Each header in the list is composed of a name and a value.""" return SESMailCommonHeaders(self["commonHeaders"])
Ancestors
Instance variables
var common_headers : SESMailCommonHeaders
-
A list of headers common to all emails. Each header in the list is composed of a name and a value.
Expand source code
@property def common_headers(self) -> SESMailCommonHeaders: """A list of headers common to all emails. Each header in the list is composed of a name and a value.""" return SESMailCommonHeaders(self["commonHeaders"])
var destination : List[str]
-
A complete list of all recipient addresses (including To: and CC: recipients) from the MIME headers of the incoming email.
Expand source code
@property def destination(self) -> List[str]: """A complete list of all recipient addresses (including To: and CC: recipients) from the MIME headers of the incoming email.""" return self["destination"]
var headers : Iterator[SESMailHeader]
-
A list of Amazon SES headers and your custom headers. Each header in the list has a name field and a value field
Expand source code
@property def headers(self) -> Iterator[SESMailHeader]: """A list of Amazon SES headers and your custom headers. Each header in the list has a name field and a value field""" for header in self["headers"]: yield SESMailHeader(header)
var headers_truncated : bool
-
String that specifies whether the headers were truncated in the notification, which will happen if the headers are larger than 10 KB. Possible values are true and false.
Expand source code
@property def headers_truncated(self) -> bool: """String that specifies whether the headers were truncated in the notification, which will happen if the headers are larger than 10 KB. Possible values are true and false.""" return bool(self["headersTruncated"])
var message_id : str
-
String that contains the unique ID assigned to the email by Amazon SES.
If the email was delivered to Amazon S3, the message ID is also the Amazon S3 object key that was used to write the message to your Amazon S3 bucket.
Expand source code
@property def message_id(self) -> str: """String that contains the unique ID assigned to the email by Amazon SES. If the email was delivered to Amazon S3, the message ID is also the Amazon S3 object key that was used to write the message to your Amazon S3 bucket.""" return self["messageId"]
var source : str
-
String that contains the email address (specifically, the envelope MAIL FROM address) that the email was sent from.
Expand source code
@property def source(self) -> str: """String that contains the email address (specifically, the envelope MAIL FROM address) that the email was sent from.""" return self["source"]
var timestamp : str
-
String that contains the time at which the email was received, in ISO8601 format.
Expand source code
@property def timestamp(self) -> str: """String that contains the time at which the email was received, in ISO8601 format.""" return self["timestamp"]
Inherited members
class SESMailCommonHeaders (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESMailCommonHeaders(DictWrapper): @property def return_path(self) -> str: """The values in the Return-Path header of the email.""" return self["returnPath"] @property def get_from(self) -> List[str]: """The values in the From header of the email.""" # Note: this name conflicts with existing python builtins return self["from"] @property def date(self) -> List[str]: """The date and time when Amazon SES received the message.""" return self["date"] @property def to(self) -> List[str]: """The values in the To header of the email.""" return self["to"] @property def message_id(self) -> str: """The ID of the original message.""" return str(self["messageId"]) @property def subject(self) -> str: """The value of the Subject header for the email.""" return str(self["subject"])
Ancestors
Instance variables
var date : List[str]
-
The date and time when Amazon SES received the message.
Expand source code
@property def date(self) -> List[str]: """The date and time when Amazon SES received the message.""" return self["date"]
var get_from : List[str]
-
The values in the From header of the email.
Expand source code
@property def get_from(self) -> List[str]: """The values in the From header of the email.""" # Note: this name conflicts with existing python builtins return self["from"]
var message_id : str
-
The ID of the original message.
Expand source code
@property def message_id(self) -> str: """The ID of the original message.""" return str(self["messageId"])
var return_path : str
-
The values in the Return-Path header of the email.
Expand source code
@property def return_path(self) -> str: """The values in the Return-Path header of the email.""" return self["returnPath"]
var subject : str
-
The value of the Subject header for the email.
Expand source code
@property def subject(self) -> str: """The value of the Subject header for the email.""" return str(self["subject"])
var to : List[str]
-
The values in the To header of the email.
Expand source code
@property def to(self) -> List[str]: """The values in the To header of the email.""" return self["to"]
Inherited members
class SESMailHeader (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESMailHeader(DictWrapper): @property def name(self) -> str: return self["name"] @property def value(self) -> str: return self["value"]
Ancestors
Instance variables
var name : str
-
Expand source code
@property def name(self) -> str: return self["name"]
var value : str
-
Expand source code
@property def value(self) -> str: return self["value"]
Inherited members
class SESMessage (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESMessage(DictWrapper): @property def mail(self) -> SESMail: return SESMail(self["ses"]["mail"]) @property def receipt(self) -> SESReceipt: return SESReceipt(self["ses"]["receipt"])
Ancestors
Instance variables
var mail : SESMail
-
Expand source code
@property def mail(self) -> SESMail: return SESMail(self["ses"]["mail"])
var receipt : SESReceipt
-
Expand source code
@property def receipt(self) -> SESReceipt: return SESReceipt(self["ses"]["receipt"])
Inherited members
class SESReceipt (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESReceipt(DictWrapper): @property def timestamp(self) -> str: """String that specifies the date and time at which the action was triggered, in ISO 8601 format.""" return self["timestamp"] @property def processing_time_millis(self) -> int: """String that specifies the period, in milliseconds, from the time Amazon SES received the message to the time it triggered the action.""" return int(self["processingTimeMillis"]) @property def recipients(self) -> List[str]: """A list of recipients (specifically, the envelope RCPT TO addresses) that were matched by the active receipt rule. The addresses listed here may differ from those listed by the destination field in the mail object.""" return self["recipients"] @property def spam_verdict(self) -> SESReceiptStatus: """Object that indicates whether the message is spam.""" return SESReceiptStatus(self["spamVerdict"]) @property def virus_verdict(self) -> SESReceiptStatus: """Object that indicates whether the message contains a virus.""" return SESReceiptStatus(self["virusVerdict"]) @property def spf_verdict(self) -> SESReceiptStatus: """Object that indicates whether the Sender Policy Framework (SPF) check passed.""" return SESReceiptStatus(self["spfVerdict"]) @property def dmarc_verdict(self) -> SESReceiptStatus: """Object that indicates whether the Domain-based Message Authentication, Reporting & Conformance (DMARC) check passed.""" return SESReceiptStatus(self["dmarcVerdict"]) @property def action(self) -> SESReceiptAction: """Object that encapsulates information about the action that was executed.""" return SESReceiptAction(self["action"])
Ancestors
Instance variables
var action : SESReceiptAction
-
Object that encapsulates information about the action that was executed.
Expand source code
@property def action(self) -> SESReceiptAction: """Object that encapsulates information about the action that was executed.""" return SESReceiptAction(self["action"])
var dmarc_verdict : SESReceiptStatus
-
Object that indicates whether the Domain-based Message Authentication, Reporting & Conformance (DMARC) check passed.
Expand source code
@property def dmarc_verdict(self) -> SESReceiptStatus: """Object that indicates whether the Domain-based Message Authentication, Reporting & Conformance (DMARC) check passed.""" return SESReceiptStatus(self["dmarcVerdict"])
var processing_time_millis : int
-
String that specifies the period, in milliseconds, from the time Amazon SES received the message to the time it triggered the action.
Expand source code
@property def processing_time_millis(self) -> int: """String that specifies the period, in milliseconds, from the time Amazon SES received the message to the time it triggered the action.""" return int(self["processingTimeMillis"])
var recipients : List[str]
-
A list of recipients (specifically, the envelope RCPT TO addresses) that were matched by the active receipt rule. The addresses listed here may differ from those listed by the destination field in the mail object.
Expand source code
@property def recipients(self) -> List[str]: """A list of recipients (specifically, the envelope RCPT TO addresses) that were matched by the active receipt rule. The addresses listed here may differ from those listed by the destination field in the mail object.""" return self["recipients"]
var spam_verdict : SESReceiptStatus
-
Object that indicates whether the message is spam.
Expand source code
@property def spam_verdict(self) -> SESReceiptStatus: """Object that indicates whether the message is spam.""" return SESReceiptStatus(self["spamVerdict"])
var spf_verdict : SESReceiptStatus
-
Object that indicates whether the Sender Policy Framework (SPF) check passed.
Expand source code
@property def spf_verdict(self) -> SESReceiptStatus: """Object that indicates whether the Sender Policy Framework (SPF) check passed.""" return SESReceiptStatus(self["spfVerdict"])
var timestamp : str
-
String that specifies the date and time at which the action was triggered, in ISO 8601 format.
Expand source code
@property def timestamp(self) -> str: """String that specifies the date and time at which the action was triggered, in ISO 8601 format.""" return self["timestamp"]
var virus_verdict : SESReceiptStatus
-
Object that indicates whether the message contains a virus.
Expand source code
@property def virus_verdict(self) -> SESReceiptStatus: """Object that indicates whether the message contains a virus.""" return SESReceiptStatus(self["virusVerdict"])
Inherited members
class SESReceiptAction (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESReceiptAction(DictWrapper): @property def get_type(self) -> str: """String that indicates the type of action that was executed. Possible values are S3, SNS, Bounce, Lambda, Stop, and WorkMail """ # Note: this name conflicts with existing python builtins return self["type"] @property def function_arn(self) -> str: """String that contains the ARN of the Lambda function that was triggered. Present only for the Lambda action type.""" return self["functionArn"] @property def invocation_type(self) -> str: """String that contains the invocation type of the Lambda function. Possible values are RequestResponse and Event. Present only for the Lambda action type.""" return self["invocationType"]
Ancestors
Instance variables
var function_arn : str
-
String that contains the ARN of the Lambda function that was triggered. Present only for the Lambda action type.
Expand source code
@property def function_arn(self) -> str: """String that contains the ARN of the Lambda function that was triggered. Present only for the Lambda action type.""" return self["functionArn"]
var get_type : str
-
String that indicates the type of action that was executed.
Possible values are S3, SNS, Bounce, Lambda, Stop, and WorkMail
Expand source code
@property def get_type(self) -> str: """String that indicates the type of action that was executed. Possible values are S3, SNS, Bounce, Lambda, Stop, and WorkMail """ # Note: this name conflicts with existing python builtins return self["type"]
var invocation_type : str
-
String that contains the invocation type of the Lambda function. Possible values are RequestResponse and Event. Present only for the Lambda action type.
Expand source code
@property def invocation_type(self) -> str: """String that contains the invocation type of the Lambda function. Possible values are RequestResponse and Event. Present only for the Lambda action type.""" return self["invocationType"]
Inherited members
class SESReceiptStatus (data: Dict[str, Any])
-
Provides a single read only access to a wrapper dict
Expand source code
class SESReceiptStatus(DictWrapper): @property def status(self) -> str: return str(self["status"])
Ancestors
Instance variables
var status : str
-
Expand source code
@property def status(self) -> str: return str(self["status"])
Inherited members