Refactor notifier and add helpers

master
blallo 2022-08-24 12:42:23 +02:00
parent 060190be61
commit b15fc9cdca
Signed by: blallo
GPG Key ID: 0CBE577C9B72DC3F
2 changed files with 65 additions and 18 deletions

View File

@ -0,0 +1,41 @@
# -*- encoding: utf-8 -*-
import logging
import typing as T
def logit(logger: logging.Logger) -> T.Callable[..., T.Any]:
def wrapper(func: T.Callable[..., T.Any]) -> T.Callable[..., T.Any]:
def inner(*args, **kwargs):
try:
value = func(*args, **kwargs)
logger.info(f"Success: {func.__name__}(args={args}, kwargs={kwargs})")
return value
except Exception as e:
logger.error(f"Failure: {e}")
return inner
return wrapper
def retry_and_log(logger: logging.Logger, retries: int):
"""
This decorator wraps a function, logs at info in case of success and logs
error in case an exception is raised.
"""
def wrapper(func: T.Callable[..., T.Any]) -> T.Callable[..., T.Any]:
def inner(*args, **kwargs):
for i in range(retries):
try:
value = func(*args, **kwargs)
logger.info(
f"Success ({i+1} tentative): {func.__name__}(args={args}, kwargs={kwargs})" # noqa: E501
)
return value
except Exception as e:
logger.error(f"Failure ({i+1} tentative): {e}")
return inner
return wrapper

View File

@ -10,27 +10,13 @@ import ssl
import typing as T
from latecomers.helpers import retry_and_log
RETRIES = 3
logger = logging.getLogger(__name__)
def retry_and_log(func: T.Callable[..., T.Any]) -> T.Callable[..., T.Any]:
"""
This decorator wraps a function, logs at info in case of success and logs
error in case an exception is raised.
"""
def inner(*args, **kwargs):
for i in range(RETRIES):
try:
func(*args, **kwargs)
logger.info(f"Success ({i+1} tentative): {func.__name__}(args={args}, kwargs={kwargs})") # noqa: E501
return
except Exception as e:
logger.error(f"Failure ({i+1} tentative): {e}")
return inner
class Notifier(object):
def __init__(
self,
@ -65,7 +51,7 @@ class Notifier(object):
self._conn.sendmail(self._from, to, email)
self._conn.close()
@retry_and_log
@retry_and_log(logger, RETRIES)
def send_result(self, to: T.List[T.Text], result: bytes) -> None:
date = datetime.now().strftime("%Y-%m-%d")
body = f"Resoconto dei voli dal sito di AdR per l'aereoporto di Ciampino in data {date}" # noqa: E501
@ -90,3 +76,23 @@ class Notifier(object):
email = message.as_string()
self.send(to, email)
@retry_and_log(logger, RETRIES)
def send_no_data(self, to: T.List[T.Text]) -> None:
date = datetime.now().strftime("%Y-%m-%d")
body = f"""Attenzione
Nessun dato è stato trovato per i voli in data {date} dal sito di AdR per
l'aereoporto di Ciampino. Questo è probabile indice di errore. Contattate
il vostro tecnico preferito.
"""
message = MIMEMultipart()
message["From"] = self._from
message["To"] = ",".join(to)
message["Subject"] = f"ATTENZIONE: [{date}] Resoconto CIA da AdR"
message.attach(MIMEText(body, "plain"))
email = message.as_string()
self.send(to, email)