Specialize EmailAlerter to use SendGrid API

1 files changed, 56 insertions, 0 deletions

diff --git a/app/SendGridAlerter.py b/app/SendGridAlerter.py
new file mode 100644
index 0000000..0509a57
--- /dev/null
+++ b/app/SendGridAlerter.py
@@ -0,0 +1,56 @@
+from typing import Any
+import requests
+import traceback
+from datetime import datetime
+from app.AggroConfig import AggroConfigSendGridAlerter
+from dataclasses import asdict
+
+
+class SendGridAlerter:
+    @staticmethod
+    def from_config(config: AggroConfigSendGridAlerter) -> "SendGridAlerter":
+        return SendGridAlerter(**asdict(config))
+
+    def __init__(
+        self,
+        api_token: str,
+        email_from: str,
+        email_to: list[str],
+    ):
+        self.email_from = email_from
+        self.email_to = email_to
+        self.api_token = api_token
+
+    def send_alert(self, text: str):
+        try:
+            now = datetime.now()
+            now_text = now.strftime("%a, %d %b %Y %H:%M:%S +0000")
+            to_lst = [{"email": email} for email in self.email_to]
+            subject = f"Aggro alert on {now_text}"
+            API_URL = "https://api.sendgrid.com/v3/mail/send"
+            data = {
+                "personalizations": [{"to": to_lst}],
+                "from": {"email": self.email_from},
+                "subject": subject,
+                "content": [
+                    {
+                        "type": "text/plain",
+                        "value": text,
+                    }
+                ],
+            }
+            r = requests.post(
+                API_URL,
+                data=data,
+                headers={
+                    "Authorization": f"Bearer {self.api_token}",
+                    "Content-Type": "application/json",
+                },
+            )
+            if r.status_code >= 400:
+                raise Exception(
+                    f"[SendGridAlerter] sending alert email via HTTP returned code {r.status_code} and body:\n{r.text}"
+                )
+
+        except:
+            traceback.print_exc()