feat: send errors via email

Author: tillywoodfield

.env.test

@@ -3,3 +3,10 @@ BUCKET_REGION="test-region"
 BUCKET_NAME="test-bucket"
 BUCKET_ACCESS_KEY_ID="test-id"
 BUCKET_ACCESS_KEY_SECRET="test-secret"
+
+NOTIFICATIONS_ENABLED=1
+NOTIFICATIONS_SMTP_HOST="localhost"
+NOTIFICATIONS_SMTP_PORT=8025
+NOTIFICATIONS_SMTP_SSL_ENABLED=0
+NOTIFICATIONS_SENDER_EMAIL="[email protected]"
+NOTIFICATIONS_RECEIVER_EMAIL="[email protected]"

README.md

@@ -44,6 +44,17 @@ For local development, create a file called `.env.local`, which will be used by
 You can change which file is loaded setting the environment variable `APP_ENV`.
 For example the tests set `APP_ENV=test`, which loads variables from `.env.test`.
 
+### Email notification environment variables
+
+To send failure notifications by email, the following environment variables must be set:
+
+- `NOTIFICATIONS_ENABLED`: 1 to enable, 0 to disable
+- `NOTIFICATIONS_SMTP_HOST`
+- `NOTIFICATIONS_SMTP_PORT`
+- `NOTIFICATIONS_SMTP_SSL_ENABLED`: 1 to enable, 0 to disable
+- `NOTIFICATIONS_SENDER_EMAIL`
+- `NOTIFICATIONS_RECEIVER_EMAIL`
+
 ### Run app
 
 ```

oc4ids_datastore_pipeline/notifications.py

@@ -0,0 +1,51 @@
+import datetime
+import logging
+import os
+import smtplib
+import ssl
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+logger = logging.getLogger(__name__)
+
+
+def _send_email(errors: list[dict[str, str]]) -> None:
+    SMTP_HOST = os.environ["NOTIFICATIONS_SMTP_HOST"]
+    SMTP_PORT = int(os.environ["NOTIFICATIONS_SMTP_PORT"])
+    SMTP_SSL_ENABLED = int(os.environ["NOTIFICATIONS_SMTP_SSL_ENABLED"])
+    SENDER_EMAIL = os.environ["NOTIFICATIONS_SENDER_EMAIL"]
+    RECEIVER_EMAIL = os.environ["NOTIFICATIONS_RECEIVER_EMAIL"]
+    logger.info(f"Sending email to {RECEIVER_EMAIL}")
+    with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
+        if SMTP_SSL_ENABLED:
+            context = ssl.create_default_context()
+            server.starttls(context=context)
+        message = MIMEMultipart()
+        message["Subject"] = "Errors in OC4IDS Datastore Pipeline run"
+        message["From"] = SENDER_EMAIL
+        message["To"] = RECEIVER_EMAIL
+
+        html = f"""\
+        <h1>Errors in OC4IDS Datastore Pipeline run</h1>
+        <p>The pipeline completed at {datetime.datetime.now(datetime.UTC)}.</p>
+        <p>Please see errors for each dataset below:</p>
+        {"".join([
+            f"""
+            <h2>{error["dataset_id"]}</h2>
+            <p>Source URL: <code>{error["source_url"]}</code></p>
+            <pre><code>{error["message"]}</code></pre>
+            """
+            for error in errors
+        ])}
+        """
+        message.attach(MIMEText(html, "html"))
+
+        server.sendmail(SENDER_EMAIL, RECEIVER_EMAIL, message.as_string())
+
+
+def send_notification(errors: list[dict[str, str]]) -> None:
+    NOTIFICATIONS_ENABLED = bool(int(os.environ.get("NOTIFICATIONS_ENABLED", "0")))
+    if NOTIFICATIONS_ENABLED:
+        _send_email(errors)
+    else:
+        logger.info("Notifications are disabled, skipping")

oc4ids_datastore_pipeline/pipeline.py

@@ -15,6 +15,7 @@
     get_dataset_ids,
     save_dataset,
 )
+from oc4ids_datastore_pipeline.notifications import send_notification
 from oc4ids_datastore_pipeline.registry import (
     fetch_registered_datasets,
     get_license_name_from_url,
@@ -165,11 +166,14 @@ def process_registry() -> None:
             process_dataset(dataset_id, url)
         except Exception as e:
             logger.warning(f"Failed to process dataset {dataset_id} with error {e}")
-            errors.append({"dataset": dataset_id, "source_url": url, "errors": str(e)})
+            errors.append(
+                {"dataset_id": dataset_id, "source_url": url, "message": str(e)}
+            )
     if errors:
         logger.error(
             f"Errors while processing registry: {json.dumps(errors, indent=4)}"
         )
+        send_notification(errors)
     logger.info("Finished processing all datasets")
 
 

tests/test_notifications.py

@@ -0,0 +1,30 @@
+from unittest.mock import MagicMock
+
+from pytest_mock import MockerFixture
+
+from oc4ids_datastore_pipeline.notifications import send_notification
+
+
+def test_send_notification(mocker: MockerFixture) -> None:
+    mock_smtp_server = MagicMock()
+    patch_smtp = mocker.patch("oc4ids_datastore_pipeline.notifications.smtplib.SMTP")
+    patch_smtp.return_value = mock_smtp_server
+
+    errors = [
+        {
+            "dataset_id": "test_dataset",
+            "source_url": "https://test_dataset.json",
+            "message": "Mocked exception",
+        }
+    ]
+    send_notification(errors)
+
+    patch_smtp.assert_called_once_with("localhost", 8025)
+    with mock_smtp_server as server:
+        server.sendmail.assert_called_once()
+        sender, receiver, message = server.sendmail.call_args[0]
+        assert sender == "[email protected]"
+        assert receiver == "[email protected]"
+        assert "test_dataset" in message
+        assert "https://test_dataset.json" in message
+        assert "Mocked exception" in message

tests/test_pipeline.py

@@ -163,5 +163,6 @@ def test_process_registry_catches_exception(mocker: MockerFixture) -> None:
         "oc4ids_datastore_pipeline.pipeline.process_dataset"
     )
     patch_process_dataset.side_effect = Exception("Mocked exception")
+    mocker.patch("oc4ids_datastore_pipeline.pipeline.send_notification")
 
     process_registry()