blallo/latecomers
1
0
Fork 0
Code Issues Pull Requests Projects Releases 6 Wiki Activity

Improve notifier connection

Browse Source
This commit is contained in:
sfigato 2022-08-25 13:11:53 +02:00
parent e9e3958d2b
commit 705432c4a5
Signed by: blallo
GPG Key ID: 0CBE577C9B72DC3F
1 changed files with 17 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
latecomers/notifier.py
View File

@ -1,4 +1,5 @@
# -*- encoding: utf-8 -*- # -*- encoding: utf-8 -*-
from contextlib import contextmanager
from datetime import datetime from datetime import datetime
from email import encoders from email import encoders
from email.mime.base import MIMEBase from email.mime.base import MIMEBase
@ -27,14 +28,7 @@ class Notifier(object):
smtp_pass: T.Text, smtp_pass: T.Text,
smtp_from: T.Optional[T.Text] = None, smtp_from: T.Optional[T.Text] = None,
) -> None: ) -> None:
context = ssl.create_default_context() self._starttls = smtp_starttls
if smtp_starttls:
self._conn = smtplib.SMTP_SSL(smtp_addr, smtp_port)
self._greet = self._conn.ehlo
else:
self._conn = smtplib.SMTP(smtp_addr, smtp_port)
self._greet = lambda: self._conn.starttls(context=context)
self._addr = smtp_addr self._addr = smtp_addr
self._port = smtp_port self._port = smtp_port
self._user = smtp_user self._user = smtp_user
@ -42,14 +36,25 @@ class Notifier(object):
self._from = smtp_from if smtp_from else smtp_user self._from = smtp_from if smtp_from else smtp_user
logger.info("%s initialized", self) logger.info("%s initialized", self)
@contextmanager
def connect(self):
context = ssl.create_default_context()
if self._starttls:
with smtplib.SMTP(self._addr, self._port) as conn:
conn.starttls(context=context)
conn.login(self._user, self._pass)
yield conn
else:
with smtplib.SMTP_SSL(self._addr, self._port) as conn:
conn.login(self._user, self._pass)
yield conn
def __str__(self) -> T.Text: def __str__(self) -> T.Text:
return f"Notifier<{self._addr}:{self._port},from={self._from}>" return f"Notifier<{self._addr}:{self._port},from={self._from}>"
def send(self, to: T.List[T.Text], email: T.Text) -> None: def send(self, to: T.List[T.Text], email: T.Text) -> None:
self._greet() with self.connect() as conn:
self._conn.login(self._user, self._pass) conn.sendmail(self._from, to, email)
self._conn.sendmail(self._from, to, email)
self._conn.close()
@retry_and_log(logger, RETRIES) @retry_and_log(logger, RETRIES)
def send_result(self, to: T.List[T.Text], result: bytes) -> None: def send_result(self, to: T.List[T.Text], result: bytes) -> None: