Merge pull request #3211 from HHS/OPS-3199/initial-message-bus

Ops 3199/initial message bus

backend/ops_api/ops/__init__.py

@@ -10,12 +10,15 @@
 from sqlalchemy import event
 from sqlalchemy.orm import Session
 
+from models import OpsEventType
 from models.utils import track_db_history_after, track_db_history_before, track_db_history_catch_errors
 from ops_api.ops.auth.decorators import check_user_session_function
 from ops_api.ops.auth.extension_config import jwtMgr
 from ops_api.ops.db import handle_create_update_by_attrs, init_db
 from ops_api.ops.error_handlers import register_error_handlers
 from ops_api.ops.home_page.views import home
+from ops_api.ops.services.can_messages import can_history_trigger
+from ops_api.ops.services.message_bus import MessageBus
 from ops_api.ops.urls import register_api
 from ops_api.ops.utils.core import is_fake_user, is_unit_test
 
@@ -24,7 +27,7 @@
 time.tzset()
 
 
-def create_app() -> Flask:
+def create_app() -> Flask:  # noqa: C901
     from ops_api.ops.utils.core import is_unit_test
 
     log_level = "INFO" if not is_unit_test() else "DEBUG"
@@ -88,6 +91,12 @@ def create_app() -> Flask:
     def shutdown_session(exception=None):
         app.db_session.remove()
 
+    @app.teardown_request
+    def teardown_request(exception=None):
+        if hasattr(request, "message_bus"):
+            request.message_bus.handle()
+            request.message_bus.cleanup()
+
     @event.listens_for(db_session, "before_commit")
     def receive_before_commit(session: Session):
         track_db_history_before(session, current_user)
@@ -162,3 +171,6 @@ def before_request_function(app: Flask, request: request):
         if not is_unit_test() and not is_fake_user(app, current_user):
             current_app.logger.info(f"Checking user session for {current_user.oidc_id}")
             check_user_session_function(current_user)
+
+    request.message_bus = MessageBus()
+    request.message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, can_history_trigger)

backend/ops_api/ops/services/can_messages.py

@@ -0,0 +1,12 @@
+from loguru import logger
+from sqlalchemy.orm import Session
+
+from models import OpsEvent
+
+
+def can_history_trigger(
+    event: OpsEvent,
+    session: Session,
+):
+    logger.debug(f"Handling event {event.event_type} with details: {event.event_details}")
+    assert session is not None

backend/ops_api/ops/services/message_bus.py

@@ -0,0 +1,65 @@
+from typing import List
+
+from blinker import signal
+from flask import current_app
+from loguru import logger
+
+from models import OpsEvent, OpsEventType
+
+
+class MessageBus:
+    """
+    A simple message bus implementation that allows for publishing and subscribing to events.
+
+    This message bus implementation uses the Blinker library to handle event signals.
+
+    This message bus assumes it exists in a single thread and is meant to be used within the context of a single request.
+
+    Published events are stored in a list and are handled when the handle method is called (usually at the end of a request).
+    """
+
+    published_events: List[OpsEvent] = []
+    known_callbacks = []
+
+    def handle(self):
+        """
+        Handle all published events by calling the appropriate handlers for each event type.
+        """
+        for event in self.published_events:
+            ops_signal = signal(event.event_type.name)
+            ops_signal.send(event, session=current_app.db_session)
+            logger.debug(f"Handling event {event}")
+        self.published_events.clear()
+
+    def subscribe(self, event_type: OpsEventType, callback: callable):
+        """
+        Subscribe to an event type with a callback function.
+
+        :param event_type: The event type to subscribe to.
+        :param callback: The callback function to call when the event is published.
+        """
+        logger.debug(f"Subscribing to {event_type} with callback {callback}")
+        ops_signal = signal(event_type.name)
+        ops_signal.connect(callback)
+
+    def publish(self, event_type: OpsEventType, event: OpsEvent):
+        """
+        Publish an event with the given event type and details.
+
+        N.B. This method does not handle the event immediately. The event will be handled when the handle method is called.
+
+        :param event_type: The event type to publish.
+        :param event: The event details to publish.
+        """
+        logger.debug(f"Publishing event {event_type} with details {event}")
+        self.published_events.append(event)
+
+    def cleanup(self):
+        """
+        Clean up all subscriptions and published events.
+        """
+        for callback in self.known_callbacks:
+            ops_signal = signal(callback)
+            ops_signal.disconnect(callback)
+        self.published_events.clear()
+        self.known_callbacks.clear()

backend/ops_api/ops/utils/events.py

@@ -3,6 +3,7 @@
 
 from flask import current_app, request
 from flask_jwt_extended import current_user
+from loguru import logger
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import UnsupportedMediaType
 
@@ -58,7 +59,11 @@ def __exit__(
             current_app.logger.info(f"EVENT: {event.to_dict()}")
 
         if isinstance(exc_val, Exception):
-            current_app.logger.error(f"EVENT ({exc_type}): {exc_val}")
+            logger.error(f"EVENT ({exc_type}): {exc_val}")
 
         if not current_app.db_session.is_active:
-            current_app.logger.error("Session is not active. It has likely been rolled back.")
+            logger.error("Session is not active. It has likely been rolled back.")
+
+        if hasattr(request, "message_bus"):
+            logger.info(f"Publishing event {self.event_type.name}")
+            request.message_bus.publish(self.event_type.name, event)

backend/ops_api/tests/ops/messagebus/test_message_bus.py

@@ -0,0 +1,50 @@
+import pytest
+
+from models import OpsEvent, OpsEventType
+from ops_api.ops.services.message_bus import MessageBus
+from ops_api.ops.utils.events import OpsEventHandler
+
+
+@pytest.mark.usefixtures("app_ctx")
+def test_message_bus_handle(loaded_db, mocker):
+    mock_callback_1 = mocker.MagicMock()
+    mock_callback_2 = mocker.MagicMock()
+    mock_callback_3 = mocker.MagicMock()
+
+    message_bus = MessageBus()
+    message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, mock_callback_1)
+    message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, mock_callback_2)
+    message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, mock_callback_3)
+
+    message_bus.publish(OpsEventType.CREATE_NEW_CAN, OpsEvent(event_type=OpsEventType.CREATE_NEW_CAN))
+
+    message_bus.handle()
+
+    mock_callback_1.assert_called()
+    mock_callback_2.assert_called()
+    mock_callback_3.assert_called()
+
+
+@pytest.mark.usefixtures("app_ctx")
+def test_message_bus_create_cans(loaded_db, mocker):
+    mock_callback_1 = mocker.MagicMock()
+    mock_callback_2 = mocker.MagicMock()
+    mock_callback_3 = mocker.MagicMock()
+
+    message_bus = MessageBus()
+
+    # patch the request object
+    r_patch = mocker.patch("ops_api.ops.utils.events.request")
+    r_patch.message_bus = message_bus
+    r_patch.message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, mock_callback_1)
+    r_patch.message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, mock_callback_2)
+    r_patch.message_bus.subscribe(OpsEventType.CREATE_NEW_CAN, mock_callback_3)
+
+    oeh = OpsEventHandler(OpsEventType.CREATE_NEW_CAN)
+    oeh.__exit__(None, None, None)
+
+    message_bus.handle()
+
+    mock_callback_1.assert_called()
+    mock_callback_2.assert_called()
+    mock_callback_3.assert_called()