aboutsummaryrefslogtreecommitdiffstats
path: root/app/SendGridAlerter.py
blob: adb2296dbeee21d730d4ecbb6529509c8d7f2ca6 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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")

            from_name, from_email = self.email_from.split(":")
            to_lst = [{"email": to} for to 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": from_email, "name": from_name},
                "subject": f"Aggro alert on {now_text}",
                "content": [
                    {
                        "type": "text/html",
                        "value": f"<pre>{text}</pre>",
                    }
                ],
            }
            r = requests.post(
                API_URL,
                json=data,
                headers={"Authorization": f"Bearer {self.api_token}"},
            )
            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()