Skip to content

Commit

Permalink
Added zepto mail support
Browse files Browse the repository at this point in the history
  • Loading branch information
devvspaces committed Aug 13, 2024
1 parent dc05634 commit 43eabe2
Showing 1 changed file with 99 additions and 0 deletions.
99 changes: 99 additions & 0 deletions src/messenger/email_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,3 +132,102 @@ def send(
logger.debug(response.content)
logger.debug(response.status_code)
return stat


class ZeptoEmailManager(BaseEmailManager):

@overload
def __init__(
self, api_key: str, sender: str, debug: bool,
block_send: bool, reply_email: str
) -> None:
...

def __init__(
self, api_key: str, *args, **kwargs
) -> None:
"""
Zepto email manager
:param api_key: api key
:type api_key: str
"""
super().__init__(*args, **kwargs)
self.__api_key = api_key
self.__headers: dict = self.set_headers()

def get_api_key(self) -> str:
"""
Get zeptomail token
:return: api key
:rtype: str
"""
return self.__api_key

def set_headers(self) -> dict:
"""
Set headers for zeptomail
:return: headers
:rtype: dict
"""
headers = {
'Authorization': self.get_api_key()
}
return headers

def get_headers(self) -> dict:
"""
Get headers
:return: headers
:rtype: dict
"""
return self.__headers

def get_post_data(
self, email: str, subject: str, message: str
) -> Dict[str, str]:
"""
Get post data for zeptomail
:param email: email of the receiver
:type email: str
:param subject: subject of the email
:type subject: str
:param message: message of the email
:type message: str
:return: post data
:rtype: Dict[str, str]
"""
data = {
"to": [{"email_address": {"address": email}}],
"from": {"address": self.get_sender()},
"subject": subject,
"htmlbody": message
}
return data

def get_post_url(self) -> str:
"""
Get post url for zepto
:return: post url
:rtype: str
"""
return 'https://api.zeptomail.com/v1.1/email'

def send(
self, receipient: str, subject: str, message: str, **kwargs
):
response = requests.post(
url=self.get_post_url(),
json=self.get_post_data(receipient, subject, message),
headers=self.get_headers()
)
stat = is_success(response.status_code)
if not stat and self.get_debug():
logger.debug(response.content)
logger.debug(response.status_code)
return stat

0 comments on commit 43eabe2

Please sign in to comment.