"""Twilio SendGrid v3/mail/send response body builder"""
from .bcc_email import Bcc
from .cc_email import Cc
from .content import Content
from .custom_arg import CustomArg
from .dynamic_template_data import DynamicTemplateData
from .email import Email
from .from_email import From
from .header import Header
from .mime_type import MimeType
from .personalization import Personalization
from .reply_to import ReplyTo
from .send_at import SendAt
from .subject import Subject
from .substitution import Substitution
from .template_id import TemplateId
from .to_email import To
class Mail(object):
"""Creates the response body for v3/mail/send"""
def __init__(
self,
from_email=None,
to_emails=None,
subject=None,
plain_text_content=None,
html_content=None,
amp_html_content=None,
global_substitutions=None,
is_multiple=False):
"""
Creates the response body for a v3/mail/send API call
:param from_email: The email address of the sender
:type from_email: From, tuple, optional
:param subject: The subject of the email
:type subject: Subject, optional
:param to_emails: The email address of the recipient
:type to_emails: To, str, tuple, list(str), list(tuple),
list(To), optional
:param plain_text_content: The plain text body of the email
:type plain_text_content: string, optional
:param html_content: The html body of the email
:type html_content: string, optional
:param amp_html_content: The amp-html body of the email
:type amp_html_content: string, optional
"""
self._attachments = None
self._categories = None
self._contents = None
self._custom_args = None
self._headers = None
self._personalizations = []
self._sections = None
self._asm = None
self._batch_id = None
self._from_email = None
self._ip_pool_name = None
self._mail_settings = None
self._reply_to = None
self._reply_to_list = None
self._send_at = None
self._subject = None
self._template_id = None
self._tracking_settings = None
# Minimum required data to send a single email
if from_email is not None:
self.from_email = from_email
if to_emails is not None:
self.add_to(to_emails, global_substitutions, is_multiple)
if subject is not None:
self.subject = subject
if plain_text_content is not None:
self.add_content(plain_text_content, MimeType.text)
if amp_html_content is not None:
self.add_content(amp_html_content, MimeType.amp)
if html_content is not None:
self.add_content(html_content, MimeType.html)
def __str__(self):
"""A JSON-ready string representation of this Mail object.
:returns: A JSON-ready string representation of this Mail object.
:rtype: string
"""
return str(self.get())
def _ensure_append(self, new_items, append_to, index=0):
"""Ensure an item is appended to a list or create a new empty list
:param new_items: the item(s) to append
:type new_items: list(obj)
:param append_to: the list on which to append the items
:type append_to: list()
:param index: index of the list on which to append the items
:type index: int
"""
append_to = append_to or []
append_to.insert(index, new_items)
return append_to
def _ensure_insert(self, new_items, insert_to):
"""Ensure an item is inserted to a list or create a new empty list
:param new_items: the item(s) to insert
:type new_items: list(obj)
:param insert_to: the list on which to insert the items at index 0
:type insert_to: list()
"""
insert_to = insert_to or []
insert_to.insert(0, new_items)
return insert_to
def _flatten_dicts(self, dicts):
"""Flatten a dict
:param dicts: Flatten a dict
:type dicts: list(dict)
"""
d = dict()
list_of_dicts = [d.get() for d in dicts or []]
return {k: v for d in list_of_dicts for k, v in d.items()}
def _get_or_none(self, from_obj):
"""Get the JSON representation of the object, else return None
:param from_obj: Get the JSON representation of the object,
else return None
:type from_obj: obj
"""
return from_obj.get() if from_obj is not None else None
def _set_emails(
self, emails, global_substitutions=None, is_multiple=False, p=0):
"""Adds emails to the Personalization object
:param emails: An Email or list of Email objects
:type emails: Email, list(Email)
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
# Send multiple emails to multiple recipients
if is_multiple is True:
if isinstance(emails, list):
for email in emails:
personalization = Personalization()
personalization.add_email(email)
self.add_personalization(personalization)
else:
personalization = Personalization()
personalization.add_email(emails)
self.add_personalization(personalization)
if global_substitutions is not None:
if isinstance(global_substitutions, list):
for substitution in global_substitutions:
for p in self.personalizations:
p.add_substitution(substitution)
else:
for p in self.personalizations:
p.add_substitution(global_substitutions)
else:
try:
personalization = self._personalizations[p]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
if isinstance(emails, list):
for email in emails:
personalization.add_email(email)
else:
personalization.add_email(emails)
if global_substitutions is not None:
if isinstance(global_substitutions, list):
for substitution in global_substitutions:
personalization.add_substitution(substitution)
else:
personalization.add_substitution(global_substitutions)
if not has_internal_personalization:
self.add_personalization(personalization, index=p)
@property
def personalizations(self):
"""A list of one or more Personalization objects
:rtype: list(Personalization)
"""
return self._personalizations
def add_personalization(self, personalization, index=0):
"""Add a Personalization object
:param personalization: Add a Personalization object
:type personalization: Personalization
:param index: The index where to add the Personalization
:type index: int
"""
self._personalizations = self._ensure_append(
personalization, self._personalizations, index)
@property
def to(self):
pass
@to.setter
def to(self, to_emails, global_substitutions=None, is_multiple=False, p=0):
"""Adds To objects to the Personalization object
:param to_emails: The email addresses of all recipients
:type to_emails: To, str, tuple, list(str), list(tuple), list(To)
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
if isinstance(to_emails, list):
for email in to_emails:
if isinstance(email, str):
email = To(email, None)
if isinstance(email, tuple):
email = To(email[0], email[1])
self.add_to(email, global_substitutions, is_multiple, p)
else:
if isinstance(to_emails, str):
to_emails = To(to_emails, None)
if isinstance(to_emails, tuple):
to_emails = To(to_emails[0], to_emails[1])
self.add_to(to_emails, global_substitutions, is_multiple, p)
def add_to(
self, to_email, global_substitutions=None, is_multiple=False, p=0):
"""Adds a To object to the Personalization object
:param to_email: A To object
:type to_email: To, str, tuple, list(str), list(tuple), list(To)
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
if isinstance(to_email, list):
for email in to_email:
if isinstance(email, str):
email = To(email, None)
elif isinstance(email, tuple):
email = To(email[0], email[1])
elif not isinstance(email, Email):
raise ValueError(
'Please use a To/Cc/Bcc, tuple, or a str for a to_email list.'
)
self._set_emails(email, global_substitutions, is_multiple, p)
else:
if isinstance(to_email, str):
to_email = To(to_email, None)
if isinstance(to_email, tuple):
to_email = To(to_email[0], to_email[1])
if isinstance(to_email, Email):
p = to_email.personalization
self._set_emails(to_email, global_substitutions, is_multiple, p)
@property
def cc(self):
pass
@cc.setter
def cc(self, cc_emails, global_substitutions=None, is_multiple=False, p=0):
"""Adds Cc objects to the Personalization object
:param cc_emails: An Cc or list of Cc objects
:type cc_emails: Cc, list(Cc), tuple
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
if isinstance(cc_emails, list):
for email in cc_emails:
if isinstance(email, str):
email = Cc(email, None)
if isinstance(email, tuple):
email = Cc(email[0], email[1])
self.add_cc(email, global_substitutions, is_multiple, p)
else:
if isinstance(cc_emails, str):
cc_emails = Cc(cc_emails, None)
if isinstance(cc_emails, tuple):
cc_emails = To(cc_emails[0], cc_emails[1])
self.add_cc(cc_emails, global_substitutions, is_multiple, p)
def add_cc(
self, cc_email, global_substitutions=None, is_multiple=False, p=0):
"""Adds a Cc object to the Personalization object
:param to_emails: An Cc object
:type to_emails: Cc
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
if isinstance(cc_email, str):
cc_email = Cc(cc_email, None)
if isinstance(cc_email, tuple):
cc_email = Cc(cc_email[0], cc_email[1])
if isinstance(cc_email, Email):
p = cc_email.personalization
self._set_emails(
cc_email, global_substitutions, is_multiple=is_multiple, p=p)
@property
def bcc(self):
pass
@bcc.setter
def bcc(
self,
bcc_emails,
global_substitutions=None,
is_multiple=False,
p=0):
"""Adds Bcc objects to the Personalization object
:param bcc_emails: An Bcc or list of Bcc objects
:type bcc_emails: Bcc, list(Bcc), tuple
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
if isinstance(bcc_emails, list):
for email in bcc_emails:
if isinstance(email, str):
email = Bcc(email, None)
if isinstance(email, tuple):
email = Bcc(email[0], email[1])
self.add_bcc(email, global_substitutions, is_multiple, p)
else:
if isinstance(bcc_emails, str):
bcc_emails = Bcc(bcc_emails, None)
if isinstance(bcc_emails, tuple):
bcc_emails = Bcc(bcc_emails[0], bcc_emails[1])
self.add_bcc(bcc_emails, global_substitutions, is_multiple, p)
def add_bcc(
self,
bcc_email,
global_substitutions=None,
is_multiple=False,
p=0):
"""Adds a Bcc object to the Personalization object
:param to_emails: An Bcc object
:type to_emails: Bcc
:param global_substitutions: A dict of substitutions for all recipients
:type global_substitutions: dict
:param is_multiple: Create a new personalization for each recipient
:type is_multiple: bool
:param p: p is the Personalization object or Personalization object
index
:type p: Personalization, integer, optional
"""
if isinstance(bcc_email, str):
bcc_email = Bcc(bcc_email, None)
if isinstance(bcc_email, tuple):
bcc_email = Bcc(bcc_email[0], bcc_email[1])
if isinstance(bcc_email, Email):
p = bcc_email.personalization
self._set_emails(
bcc_email,
global_substitutions,
is_multiple=is_multiple,
p=p)
@property
def subject(self):
"""The global Subject object
:rtype: Subject
"""
return self._subject
@subject.setter
def subject(self, value):
"""The subject of the email(s)
:param value: The subject of the email(s)
:type value: Subject, string
"""
if isinstance(value, Subject):
if value.personalization is not None:
try:
personalization = \
self._personalizations[value.personalization]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
personalization.subject = value.subject
if not has_internal_personalization:
self.add_personalization(
personalization,
index=value.personalization)
else:
self._subject = value
else:
self._subject = Subject(value)
@property
def headers(self):
"""A list of global Header objects
:rtype: list(Header)
"""
return self._headers
@property
def header(self):
pass
@header.setter
def header(self, headers):
"""Add headers to the email
:param value: A list of Header objects or a dict of header key/values
:type value: Header, list(Header), dict
"""
if isinstance(headers, list):
for h in headers:
self.add_header(h)
else:
self.add_header(headers)
def add_header(self, header):
"""Add headers to the email globaly or to a specific Personalization
:param value: A Header object or a dict of header key/values
:type value: Header, dict
"""
if header.personalization is not None:
try:
personalization = \
self._personalizations[header.personalization]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
if isinstance(header, dict):
(k, v) = list(header.items())[0]
personalization.add_header(Header(k, v))
else:
personalization.add_header(header)
if not has_internal_personalization:
self.add_personalization(
personalization,
index=header.personalization)
else:
if isinstance(header, dict):
(k, v) = list(header.items())[0]
self._headers = self._ensure_append(
Header(k, v), self._headers)
else:
self._headers = self._ensure_append(header, self._headers)
@property
def substitution(self):
pass
@substitution.setter
def substitution(self, substitution):
"""Add substitutions to the email
:param value: Add substitutions to the email
:type value: Substitution, list(Substitution)
"""
if isinstance(substitution, list):
for s in substitution:
self.add_substitution(s)
else:
self.add_substitution(substitution)
def add_substitution(self, substitution):
"""Add a substitution to the email
:param value: Add a substitution to the email
:type value: Substitution
"""
if substitution.personalization:
try:
personalization = \
self._personalizations[substitution.personalization]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
personalization.add_substitution(substitution)
if not has_internal_personalization:
self.add_personalization(
personalization, index=substitution.personalization)
else:
if isinstance(substitution, list):
for s in substitution:
for p in self.personalizations:
p.add_substitution(s)
else:
for p in self.personalizations:
p.add_substitution(substitution)
@property
def custom_args(self):
"""A list of global CustomArg objects
:rtype: list(CustomArg)
"""
return self._custom_args
@property
def custom_arg(self):
return self._custom_args
@custom_arg.setter
def custom_arg(self, custom_arg):
"""Add custom args to the email
:param value: A list of CustomArg objects or a dict of custom arg
key/values
:type value: CustomArg, list(CustomArg), dict
"""
if isinstance(custom_arg, list):
for c in custom_arg:
self.add_custom_arg(c)
else:
self.add_custom_arg(custom_arg)
def add_custom_arg(self, custom_arg):
"""Add custom args to the email globaly or to a specific Personalization
:param value: A CustomArg object or a dict of custom arg key/values
:type value: CustomArg, dict
"""
if custom_arg.personalization is not None:
try:
personalization = \
self._personalizations[custom_arg.personalization]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
if isinstance(custom_arg, dict):
(k, v) = list(custom_arg.items())[0]
personalization.add_custom_arg(CustomArg(k, v))
else:
personalization.add_custom_arg(custom_arg)
if not has_internal_personalization:
self.add_personalization(
personalization, index=custom_arg.personalization)
else:
if isinstance(custom_arg, dict):
(k, v) = list(custom_arg.items())[0]
self._custom_args = self._ensure_append(
CustomArg(k, v), self._custom_args)
else:
self._custom_args = self._ensure_append(
custom_arg, self._custom_args)
@property
def send_at(self):
"""The global SendAt object
:rtype: SendAt
"""
return self._send_at
@send_at.setter
def send_at(self, value):
"""A unix timestamp specifying when your email should
be delivered.
:param value: A unix timestamp specifying when your email should
be delivered.
:type value: SendAt, int
"""
if isinstance(value, SendAt):
if value.personalization is not None:
try:
personalization = \
self._personalizations[value.personalization]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
personalization.send_at = value.send_at
if not has_internal_personalization:
self.add_personalization(
personalization, index=value.personalization)
else:
self._send_at = value
else:
self._send_at = SendAt(value)
@property
def dynamic_template_data(self):
pass
@dynamic_template_data.setter
def dynamic_template_data(self, value):
"""Data for a transactional template
:param value: Data for a transactional template
:type value: DynamicTemplateData, a JSON-serializable structure
"""
if not isinstance(value, DynamicTemplateData):
value = DynamicTemplateData(value)
try:
personalization = self._personalizations[value.personalization]
has_internal_personalization = True
except IndexError:
personalization = Personalization()
has_internal_personalization = False
personalization.dynamic_template_data = value.dynamic_template_data
if not has_internal_personalization:
self.add_personalization(
personalization, index=value.personalization)
@property
def from_email(self):
"""The email address of the sender
:rtype: From
"""
return self._from_email
@from_email.setter
def from_email(self, value):
"""The email address of the sender
:param value: The email address of the sender
:type value: From, str, tuple
"""
if isinstance(value, str):
value = From(value, None)
if isinstance(value, tuple):
value = From(value[0], value[1])
self._from_email = value
@property
def reply_to(self):
"""The reply to email address
:rtype: ReplyTo
"""
return self._reply_to
@reply_to.setter
def reply_to(self, value):
"""The reply to email address
:param value: The reply to email address
:type value: ReplyTo, str, tuple
"""
if isinstance(value, str):
value = ReplyTo(value, None)
if isinstance(value, tuple):
value = ReplyTo(value[0], value[1])
self._reply_to = value
@property
def reply_to_list(self):
"""A list of ReplyTo email addresses
:rtype: list(ReplyTo), tuple
"""
return self._reply_to_list
@reply_to_list.setter
def reply_to_list(self, value):
"""A list of ReplyTo email addresses
:param value: A list of ReplyTo email addresses
:type value: list(ReplyTo), tuple
"""
if isinstance(value, list):
for reply in value:
if isinstance(reply, ReplyTo):
if not isinstance(reply.email, str):
raise ValueError('You must provide an email for each entry in a reply_to_list')
else:
raise ValueError(
'Please use a list of ReplyTos for a reply_to_list.'
)
self._reply_to_list = value
@property
def contents(self):
"""The contents of the email
:rtype: list(Content)
"""
return self._contents
@property
def content(self):
pass
@content.setter
def content(self, contents):
"""The content(s) of the email
:param contents: The content(s) of the email
:type contents: Content, list(Content)
"""
if isinstance(contents, list):
for c in contents:
self.add_content(c)
else:
self.add_content(contents)
def add_content(self, content, mime_type=None):
"""Add content to the email
:param contents: Content to be added to the email
:type contents: Content
:param mime_type: Override the mime type
:type mime_type: MimeType, str
"""
if isinstance(content, str):
content = Content(mime_type, content)
# Content of mime type text/plain must always come first, followed by text/x-amp-html and then text/html
if content.mime_type == MimeType.text:
self._contents = self._ensure_insert(content, self._contents)
elif content.mime_type == MimeType.amp:
if self._contents:
for _content in self._contents:
# this is written in the context that plain text content will always come earlier than the html content
if _content.mime_type == MimeType.text:
index = 1
break
elif _content.mime_type == MimeType.html:
index = 0
break
else:
index = 0
self._contents = self._ensure_append(
content, self._contents, index=index)
else:
if self._contents:
index = len(self._contents)
else:
index = 0
self._contents = self._ensure_append(
content, self._contents, index=index)
@property
def attachments(self):
"""The attachments to this email
:rtype: list(Attachment)
"""
return self._attachments
@property
def attachment(self):
pass
@attachment.setter
def attachment(self, attachment):
"""Add attachment(s) to this email
:param attachment: Add attachment(s) to this email
:type attachment: Attachment, list(Attachment)
"""
if isinstance(attachment, list):
for a in attachment:
self.add_attachment(a)
else:
self.add_attachment(attachment)
def add_attachment(self, attachment):
"""Add an attachment to this email
:param attachment: Add an attachment to this email
:type attachment: Attachment
"""
self._attachments = self._ensure_append(attachment, self._attachments)
@property
def template_id(self):
"""The transactional template id for this email
:rtype: TemplateId
"""
return self._template_id
@template_id.setter
def template_id(self, value):
"""The transactional template id for this email
:param value: The transactional template id for this email
:type value: TemplateId
"""
if isinstance(value, TemplateId):
self._template_id = value
else:
self._template_id = TemplateId(value)
@property
def sections(self):
"""The block sections of code to be used as substitutions
:rtype: Section
"""
return self._sections
@property
def section(self):
pass
@section.setter
def section(self, section):
"""The block sections of code to be used as substitutions
:rtype: Section, list(Section)
"""
if isinstance(section, list):
for h in section:
self.add_section(h)
else:
self.add_section(section)
def add_section(self, section):
"""A block section of code to be used as substitutions
:param section: A block section of code to be used as substitutions
:type section: Section
"""
self._sections = self._ensure_append(section, self._sections)
@property
def categories(self):
"""The categories assigned to this message
:rtype: list(Category)
"""
return self._categories
@property
def category(self):
pass
@category.setter
def category(self, categories):
"""Add categories assigned to this message
:rtype: list(Category)
"""
if isinstance(categories, list):
for c in categories:
self.add_category(c)
else:
self.add_category(categories)
def add_category(self, category):
"""Add a category assigned to this message
:rtype: Category
"""
self._categories = self._ensure_append(category, self._categories)
@property
def batch_id(self):
"""The batch id for this email
:rtype: BatchId
"""
return self._batch_id
@batch_id.setter
def batch_id(self, value):
"""The batch id for this email
:param value: The batch id for this email
:type value: BatchId
"""
self._batch_id = value
@property
def asm(self):
"""An object specifying unsubscribe behavior.
:rtype: Asm
"""
return self._asm
@asm.setter
def asm(self, value):
"""An object specifying unsubscribe behavior.
:param value: An object specifying unsubscribe behavior.
:type value: Asm
"""
self._asm = value
@property
def ip_pool_name(self):
"""The IP Pool that you would like to send this email from
:rtype: IpPoolName
"""
return self._ip_pool_name
@ip_pool_name.setter
def ip_pool_name(self, value):
"""The IP Pool that you would like to send this email from
:paran value: The IP Pool that you would like to send this email from
:type value: IpPoolName
"""
self._ip_pool_name = value
@property
def mail_settings(self):
"""The mail settings for this email
:rtype: MailSettings
"""
return self._mail_settings
@mail_settings.setter
def mail_settings(self, value):
"""The mail settings for this email
:param value: The mail settings for this email
:type value: MailSettings
"""
self._mail_settings = value
@property
def tracking_settings(self):
"""The tracking settings for this email
:rtype: TrackingSettings
"""
return self._tracking_settings
@tracking_settings.setter
def tracking_settings(self, value):
"""The tracking settings for this email
:param value: The tracking settings for this email
:type value: TrackingSettings
"""
self._tracking_settings = value
def get(self):
"""
Get a JSON-ready representation of this Mail object.
:returns: This Mail object, ready for use in a request body.
:rtype: dict
"""
mail = {
'from': self._get_or_none(self.from_email),
'subject': self._get_or_none(self.subject),
'personalizations': [p.get() for p in self.personalizations or []],
'content': [c.get() for c in self.contents or []],
'attachments': [a.get() for a in self.attachments or []],
'template_id': self._get_or_none(self.template_id),
'sections': self._flatten_dicts(self.sections),
'headers': self._flatten_dicts(self.headers),
'categories': [c.get() for c in self.categories or []],
'custom_args': self._flatten_dicts(self.custom_args),
'send_at': self._get_or_none(self.send_at),
'batch_id': self._get_or_none(self.batch_id),
'asm': self._get_or_none(self.asm),
'ip_pool_name': self._get_or_none(self.ip_pool_name),
'mail_settings': self._get_or_none(self.mail_settings),
'tracking_settings': self._get_or_none(self.tracking_settings),
'reply_to': self._get_or_none(self.reply_to),
'reply_to_list': [r.get() for r in self.reply_to_list or []],
}
return {key: value for key, value in mail.items()
if value is not None and value != [] and value != {}}
@classmethod
def from_EmailMessage(cls, message):
"""Create a Mail object from an instance of
email.message.EmailMessage.
:type message: email.message.EmailMessage
:rtype: Mail
"""
mail = cls(
from_email=Email(message.get('From')),
subject=message.get('Subject'),
to_emails=Email(message.get('To')),
)
try:
body = message.get_content()
except AttributeError:
# Python2
body = message.get_payload()
mail.add_content(Content(
message.get_content_type(),
body.strip()
))
for k, v in message.items():
mail.add_header(Header(k, v))
return mail
