feat(email_alerting_code): Alert users when restrictions change (#32)

modules/api/email_alerting_code/package/index.py

@@ -1,21 +1,35 @@
-import mailjet_rest
-import db_connection
+import logging
 from mailjet_rest import Client
 import os
+import psycopg2
+from psycopg2.extras import DictCursor
+
+import db_connection
+
+import repositories
+
 
 # lambda environment variables
 SECRET_NAME = os.environ['secret_name']
 REGION_NAME = os.environ['region_name']
 DB = os.environ['db']
 
-credential = db_connection.get_credentials(SECRET_NAME, REGION_NAME, DB)
+# def connect_to_local_db():
+#     connection = psycopg2.connect(database="ecowater", user="ecowater", password="ecowater", host="pg", port="5432")
+#     return connection
 
+# def get_credentials():
+#     '''get the DB credentials from Secrets Manager'''
+#     return {
+#         'mailjet_api_key': 'find key on aws',
+#         'mailjet_api_secret': 'find key on aws',
+#     }
 
-def lambda_handler(event, context):
-#def send_email(destinataire,niveau):
+credential = db_connection.get_credentials(SECRET_NAME, REGION_NAME, DB)
+# credential = get_credentials()
 
-    mailjet = Client(auth=(credential['mailjet_api_key'], credential['mailjet_api_secret']), version='v3.1')
-    data = {
+def create_email(email_address):
+    return {
     'Messages': [
             {
                 "From": {
@@ -24,22 +38,59 @@ def lambda_handler(event, context):
                 },
                 "To": [
                     {
-                        "Email": "",
-                        "Name": "passenger 1"
+                        "Email": email_address,
+                        "Name": email_address
                     }
                 ],
                 "TemplateID": 4732985,
                 "TemplateLanguage": True,
                 "Subject": "renforcée",
                 "Variables": {
-        "niveau": "maximale"
-    }
+                    "niveau": "maximale"
+                }
             }
         ]
     }
 
-    result = mailjet.send.create(data=data)
-    print (result.status_code)
-    print (result.json())
 
-    return None
+def lambda_handler(event, context):
+    logging.basicConfig(level = logging.DEBUG, format = '%(asctime)s - %(levelname)s - %(message)s')
+    logging.debug('Connected to the database')
+
+    connection = db_connection.connect_to_db(SECRET_NAME, REGION_NAME, DB)
+    # connection = connect_to_local_db()
+    connection.autocommit = True
+    cursor = connection.cursor(cursor_factory=DictCursor)
+
+    mailjet = Client(auth=(credential['mailjet_api_key'], credential['mailjet_api_secret']), version='v3.1')
+
+    # on récupère les events qui doivent être consommés
+    events = repositories.find_all_decree_events(cursor)
+
+    for event in events:
+        # on récupère le decree associé et sa geozone
+        decree_id = event[1]
+        decree = repositories.get_decree_by_id(cursor, decree_id)
+        if decree == None:
+            logging.error('Decree with id %s not found.', decree_id)
+            continue
+
+        # on récupère les utilisateurs associés à la zone
+        geozone_id = decree.get_geozone_id()
+        email_addresses = repositories.find_user_emails_by_geozone_id(cursor, geozone_id)
+
+        # on envoie les emails à chaque utilisateur
+        for email_address in email_addresses:
+            email = create_email(email_address[0])
+            result = mailjet.send.create(data=email)
+            logging.info(f"Sending email to {email_address[0]}")
+            if (result.status != 200):
+                logging.error(f"Failed to send email to {email_address[0]} (decree_id: {decree})")
+
+        # on supprime l'event
+        repositories.delete_event(cursor, event[0])
+
+    return
+
+if __name__ == '__main__':
+    lambda_handler(None, None)

modules/api/email_alerting_code/package/models.py

@@ -0,0 +1,21 @@
+from datetime import datetime
+from typing import Optional, Any
+from enum import Enum
+
+class Event(Enum):
+    DECREE_CREATION = 'decree_creation'
+    DECREE_REPEAL = 'decree_repeal'
+
+
+class Decree:
+    def __init__(self, id: Optional[int], external_id: str, geozone_id: int, alert_level: str, start_date: datetime, end_date: datetime, document: str):
+        self.id = id
+        self.external_id = external_id
+        self.geozone_id = geozone_id
+        self.alert_level = alert_level
+        self.start_date = start_date
+        self.end_date = end_date
+        self.document = document
+
+    def get_geozone_id(self):
+        return self.geozone_id

modules/api/email_alerting_code/package/repositories.py

@@ -0,0 +1,52 @@
+from typing import List,  Optional
+from models import Decree, Event
+
+
+def find_all_decree_events(cursor) -> List[Event]:
+    cursor.execute("SELECT * FROM event_store WHERE type IN ('decree_creation', 'decree_repeal');")
+    results = cursor.fetchall()
+    return [] if not results else results
+
+def get_decree_by_id(cursor, id: str) -> Optional[Decree]:
+    query = 'SELECT * FROM decree WHERE id = %s'
+    parameters = (id,)
+    cursor.execute(query, parameters)
+
+    result = cursor.fetchone()
+    if not result:
+        return None
+
+    return Decree(
+        id=result.get('id'),
+        external_id=result.get('external_id'),
+        geozone_id=result.get('geozone_id'),
+        alert_level=result.get('alert_level'),
+        start_date=result.get('start_date'),
+        end_date=result.get('end_date'),
+        document=result.get('document')
+    )
+
+
+def find_user_emails_by_geozone_id(cursor, geozone_id: str) -> List[str]:
+    query = """
+        SELECT DISTINCT email
+        FROM alert_subscription
+        CROSS JOIN geozone
+        WHERE geozone.id = %s
+        AND ST_Contains(
+	        geozone.geom,
+	        ST_SetSRID(
+		        ST_MakePoint(alert_subscription.longitude, alert_subscription.latitude),
+		        4326
+	        )
+        );
+    """
+    params = (geozone_id,)
+    cursor.execute(query, params)
+    emails = cursor.fetchall()
+    return emails
+
+def delete_event(cursor, event_id: int):
+    query = 'DELETE FROM event_store WHERE id = %s'
+    parameters = (event_id,)
+    cursor.execute(query, parameters)

modules/api/sync/package/requirements.txt

@@ -2,3 +2,4 @@ psycopg2
 requests
 email-validator
 geopandas
+mailjet-rest