Merge pull request #10 from sam-atkins/feature/add-email-notifications

Feature/add email notifications

README.md

@@ -4,7 +4,7 @@ This is a proof of concept benchmarking app/tool with an event-driven architectu
 
 - [slowking](#slowking)
   - [Design Principles](#design-principles)
-  - [Setup](#setup)
+  - [Local Dev](#local-dev)
   - [Database Migrations](#database-migrations)
   - [Design Notes](#design-notes)
     - [Commands](#commands)
@@ -28,18 +28,24 @@ This is a proof of concept benchmarking app/tool with an event-driven architectu
 - No infrastructure provisioning. It relies on a benchmarkable Platform to already be running
 - Benchmark reports will be in CSV format
 
-## Setup
+## Local Dev
 
 ```shell
 task build && task up
 ```
 
+Following are available:
+
+- Swagger Docs for the Slowking API: http://0.0.0.0:8091/docs
+- MailHog UI: http://0.0.0.0:18025/
+
 There are also various tasks in the Taskfile:
 
 ```shell
 task --list
 ```
 
+
 ## Database Migrations
 
 Alembic is used for migrations. To create a migration run this command in the activated venv.

docker-compose.yml

@@ -36,6 +36,7 @@ services:
     depends_on:
       - slowking-postgres
       - slowking-redis
+      - mailhog
     volumes:
       - ./slowking:/home/app/slowking
       - ./reports:/home/app/reports
@@ -81,6 +82,13 @@ services:
     volumes:
       - ./eigenapi:/home/app/src
 
+  mailhog:
+    image: mailhog/mailhog:v1.0.1
+    container_name: slowking-mailhog
+    ports:
+      - "11025:1025"
+      - "18025:8025"
+
 # networks:
 #   network_eigen_dev:
 #     name: eigen_dev_default

slowking/adapters/notifications.py

@@ -1,21 +1,44 @@
 """
-Notification adapters. More a placeholder than anything else, just logs right now but
-could be extended to send emails, Slack messages, etc.
+Notification adapters. Small examples for logging and email but
+could be extended to send different emails or Slack messages, etc.
 """
 import abc
 import logging.config
+import smtplib
 
-from slowking.domain import events
+from slowking.config import settings
+from slowking.domain import model
 
 logger = logging.getLogger(__name__)
 
 
 class AbstractNotifications(abc.ABC):
     @abc.abstractmethod
-    def send(self, event: events.Event, message: str):
+    def send(self, benchmark: model.Benchmark, message: str):
         raise NotImplementedError
 
 
 class LogNotifications(AbstractNotifications):
-    def send(self, event: events.Event, message: str):
-        logger.info(f"Sending notification for event {event} with message: {message}")
+    def send(self, benchmark: model.Benchmark, message: str):
+        logger.info(f"Sending notification for benchmark {benchmark.name}")
+
+
+class EmailNotifications(AbstractNotifications):
+    def __init__(self, smtp_host=settings.EMAIL_HOST, port=settings.EMAIL_PORT):
+        self.server = smtplib.SMTP(smtp_host, port=port)
+        self.server.noop()
+
+    def send(self, benchmark: model.Benchmark, message: str) -> None:
+        subject = f"Benchmark Report: {benchmark.name}"
+        body = f"Report generated: {message}"
+        msg = f"Subject: {subject}\n{body}"
+
+        # Hard coded as an example, could be taken from the benchmark aggregate
+        # the CreateBenchmark command could have an email field
+        destination = ["hello@example.com"]
+
+        self.server.sendmail(
+            from_addr="benchmarks@slowking.com",
+            to_addrs=destination,
+            msg=msg,
+        )

slowking/adapters/report.py

@@ -13,7 +13,7 @@
 
 class AbstractReporter:
     @abc.abstractmethod
-    def create(self, benchmark: model.Benchmark):
+    def create(self, benchmark: model.Benchmark) -> str:
         raise NotImplementedError
 
 
@@ -26,7 +26,7 @@ class LatencyReport(AbstractReporter):
     output_filename: str = settings.OUTPUT_FILENAME
 
     @classmethod
-    def create(cls, benchmark: model.Benchmark):
+    def create(cls, benchmark: model.Benchmark) -> str:
         """
         Create report for given benchmark.
         """
@@ -51,7 +51,8 @@ def create(cls, benchmark: model.Benchmark):
                 "Upload Time (seconds)": doc.upload_time,
             }
             fields.append({**base_info, **doc_info})
-        with open(f"{cls.output_dir}{cls.output_filename}", "w") as csv_file:
+        file_path = f"{cls.output_dir}{cls.output_filename}"
+        with open(file_path, "w") as csv_file:
             csv_writer = csv.DictWriter(
                 csv_file,
                 delimiter=",",
@@ -60,4 +61,5 @@ def create(cls, benchmark: model.Benchmark):
             )
             csv_writer.writeheader()
             csv_writer.writerows(fields)
-        logger.info("=== LatencyReport created ===")
+        logger.info(f"=== LatencyReport {file_path} created ===")
+        return file_path

slowking/bootstrap.py

@@ -3,7 +3,10 @@
 
 from slowking.adapters import orm, redis_event_publisher
 from slowking.adapters.http import EigenClient
-from slowking.adapters.notifications import AbstractNotifications, LogNotifications
+from slowking.adapters.notifications import (
+    AbstractNotifications,
+    EmailNotifications,
+)
 from slowking.domain import commands, events
 from slowking.service_layer import handlers, messagebus, unit_of_work
 
@@ -22,7 +25,7 @@ def bootstrap(
         logger.info("Bootstrap DB and ORM setup completed")
 
     if notifications is None:
-        notifications = LogNotifications()
+        notifications = EmailNotifications()
 
     injected_command_handlers: dict[Type[commands.Command], list[Callable]] = {
         commands.CreateBenchmark: [
@@ -48,7 +51,7 @@ def bootstrap(
             lambda e: handlers.check_all_documents_uploaded(e, uow, publish),
         ],
         events.AllDocumentsUploaded: [
-            lambda e: handlers.create_report(e, uow),
+            lambda e: handlers.create_report(e, uow, notifications),
         ],
     }
 

slowking/config.py

@@ -15,6 +15,9 @@ class Settings(BaseSettings):
     API_BENCHMARK_NAMESPACE_V1_STR: str = f"{API_V1_STR}/benchmarks"
     DB_MAX_RETRIES: int = 10
     DB_RETRY_INTERVAL: int = 1
+    EMAIL_HOST: str = "slowking-mailhog"
+    EMAIL_PORT: int = 1025
+    EMAIL_HTTP_PORT: int = 8025
     OUTPUT_DIR: str = "/home/app/reports/"
     OUTPUT_FILENAME: str = (
         f"report_{datetime.now(timezone.utc).strftime('%Y_%m_%d__%H_%M_%S')}.csv"

slowking/entrypoints/http.py

@@ -15,8 +15,6 @@
 router = APIRouter(prefix=settings.API_BENCHMARK_NAMESPACE_V1_STR, tags=["benchmarks"])
 
 
-bus = bootstrap.bootstrap()
-
 logging.config.dictConfig(config.logger_dict_config())
 logger = getLogger(__name__)
 
@@ -25,6 +23,7 @@ def publish_to_bus(cmd: commands.Command):
     """
     Publishes a command to the eventbus
     """
+    bus = bootstrap.bootstrap()
     bus.handle(cmd)
 
 

slowking/service_layer/handlers.py

@@ -7,6 +7,7 @@
 from sqlalchemy.exc import IllegalStateChangeError, InvalidRequestError
 from sqlalchemy.orm.exc import DetachedInstanceError
 
+from slowking.adapters import notifications
 from slowking.adapters.http import EigenClient
 from slowking.adapters.report import LatencyReport
 from slowking.config import settings
@@ -221,11 +222,14 @@ def check_all_documents_uploaded(
 
 
 def create_report(
-    event: events.AllDocumentsUploaded, uow: unit_of_work.AbstractUnitOfWork
+    event: events.AllDocumentsUploaded,
+    uow: unit_of_work.AbstractUnitOfWork,
+    notifications: notifications.AbstractNotifications,
 ):
     logger.info(f"=== Called create_report with {event} ===")
     with uow:
         bm = uow.benchmarks.get_by_id(event.benchmark_id)
         logger.info(f"=== create_report bm === : {bm}")
-        LatencyReport().create(bm)
-        # TODO notify user of report generation e.g. email with report
+        report = LatencyReport().create(bm)
+        notifications.send(benchmark=bm, message=report)
+        logger.info("=== Create Report Notification sent  ===")

tests/conftest.py

@@ -1,11 +1,18 @@
 import pytest
 from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import clear_mappers, sessionmaker
 
-from slowking.adapters.orm import metadata
+from slowking.adapters.orm import metadata, start_mappers
 from slowking.domain import model
 
 
+@pytest.fixture
+def mappers():
+    start_mappers()
+    yield
+    clear_mappers()
+
+
 @pytest.fixture
 def in_memory_sqlite_db():
     engine = create_engine("sqlite:///:memory:")

tests/integration/slowking/adapters/test_repository.py

@@ -3,8 +3,12 @@
 but do not pass when running just this module.
 The error is `UnmappedInstanceError`.
 """
+import pytest
+
 from slowking.adapters import repository
 
+pytestmark = pytest.mark.usefixtures("mappers")
+
 
 def test_get_by_id(sqlite_session_factory, benchmark):
     session = sqlite_session_factory()

tests/integration/slowking/service_layer/test_handlers.py

@@ -2,11 +2,10 @@
 from collections import defaultdict
 from datetime import datetime, timezone
 
-
 from slowking import bootstrap
 from slowking.adapters import notifications, repository
 from slowking.adapters.http import EigenClient, ProjectStruct
-from slowking.domain import commands, events
+from slowking.domain import commands, events, model
 from slowking.service_layer import unit_of_work
 
 
@@ -55,7 +54,8 @@ class FakeNotifications(notifications.AbstractNotifications):
     def __init__(self):
         self.sent = defaultdict(list)
 
-    def send(self, destination, message):
+    def send(self, benchmark: model.Benchmark, message: str):
+        destination = ["test@example.com"]
         self.sent[destination].append(message)
 
 

tests/integration/slowking/service_layer/test_unit_of_work.py

@@ -4,6 +4,8 @@
 from slowking.domain import model
 from slowking.service_layer import unit_of_work
 
+pytestmark = pytest.mark.usefixtures("mappers")
+
 
 def get_benchmark_by_name(session, name):
     benchmark = session.query(model.Benchmark).filter_by(name=name).first()