feat: enable producing to event bus via settings

docs: update adding event to bus document

chore: fix lint issues

fix: docs and type checking

Co-authored-by: Arunmozhi <[email protected]>

docs: update events how to

fix: Update openedx_events/apps.py

Co-authored-by: Arunmozhi <[email protected]>

refactor: catch bad event type and log

CHANGELOG.rst

@@ -17,6 +17,12 @@ Changed
 ~~~~~~~
 * Re-licensed this repository from AGPL 3.0 to Apache 2.0
 
+[8.6.0] - 2023-08-28
+--------------------
+Added
+~~~~~
+* Added generic handler to allow producing to event bus via django settings.
+
 [8.5.0] - 2023-08-08
 --------------------
 Changed

docs/how-tos/adding-events-to-event-bus.rst

@@ -7,8 +7,35 @@ to use the Open edX Event Bus. Here, we list useful information about
 adding a new event to the event bus:
 
 - `How to start using the Event Bus`_
-- `Sample pull request adding new Open edX Events to the Event Bus`_
 
 
 .. _How to start using the Event Bus: https://openedx.atlassian.net/wiki/spaces/AC/pages/3508699151/How+to+start+using+the+Event+Bus
-.. _Sample pull request adding new Open edX Events to the Event Bus: https://github.com/openedx/edx-platform/pull/31350
+
+
+Producing to event bus
+^^^^^^^^^^^^^^^^^^^^^^
+
+In the producing/host application, include ``openedx_events`` in ``INSTALLED_APPS`` settings and add ``EVENT_BUS_PRODUCER_CONFIG`` setting. For example, below snippet is to push ``XBLOCK_PUBLISHED`` to two different topics and ``XBLOCK_DELETED`` signal to one topic in event bus.
+
+.. code-block:: python
+
+   # .. setting_name: EVENT_BUS_PRODUCER_CONFIG
+   # .. setting_default: {}
+   # .. setting_description: Dictionary of event_types mapped to lists of dictionaries containing topic related configuration.
+   #    Each topic configuration dictionary contains
+   #    * a topic/stream name called `topic` where the event will be pushed to.
+   #    * a flag called `enabled` denoting whether the event will be published to the topic.
+   #    * `event_key_field` which is a period-delimited string path to event data field to use as event key.
+   #    Note: The topic names should not include environment prefix as it will be dynamically added based on
+   #    EVENT_BUS_TOPIC_PREFIX setting.
+   EVENT_BUS_PRODUCER_CONFIG = {
+       'org.openedx.content_authoring.xblock.published.v1': [
+           {'topic': 'content-authoring-xblock-lifecycle', 'event_key_field': 'xblock_info.usage_key', 'enabled': True},
+           {'topic': 'content-authoring-xblock-published', 'event_key_field': 'xblock_info.usage_key', 'enabled': True},
+       ],
+       'org.openedx.content_authoring.xblock.deleted.v1': [
+           {'topic': 'content-authoring-xblock-lifecycle', 'event_key_field': 'xblock_info.usage_key', 'enabled': True},
+       ],
+   }
+
+The ``EVENT_BUS_PRODUCER_CONFIG`` is read by openedx_events and a handler is attached which does the leg work of reading the configuration again and pushing to appropriate handlers.

openedx_events/__init__.py

@@ -5,4 +5,4 @@
 more information about the project.
 """
 
-__version__ = "8.5.0"
+__version__ = "8.6.0"

openedx_events/apps.py

@@ -3,11 +3,89 @@
 """
 
 from django.apps import AppConfig
+from django.conf import settings
+
+from openedx_events.event_bus import get_producer
+from openedx_events.exceptions import ProducerConfigurationError
+from openedx_events.tooling import OpenEdxPublicSignal, load_all_signals
+
+
+def general_signal_handler(sender, signal, **kwargs):  # pylint: disable=unused-argument
+    """
+    Signal handler for publishing events to configured event bus.
+    """
+    configurations = getattr(settings, "EVENT_BUS_PRODUCER_CONFIG", {}).get(signal.event_type, ())
+    event_data = {key: kwargs.get(key) for key in signal.init_data}
+    for configuration in configurations:
+        if configuration["enabled"]:
+            get_producer().send(
+                signal=signal,
+                topic=configuration["topic"],
+                event_key_field=configuration["event_key_field"],
+                event_data=event_data,
+                event_metadata=kwargs["metadata"],
+            )
 
 
 class OpenedxEventsConfig(AppConfig):
     """
     Configuration for the openedx_events Django application.
     """
 
-    name = 'openedx_events'
+    name = "openedx_events"
+
+    def _get_validated_signal_config(self, event_type, configurations):
+        """
+        Validate signal configuration format.
+
+        Raises:
+            ProducerConfigurationError: If configuration is not valid.
+        """
+        if not isinstance(configurations, list) and not isinstance(configurations, tuple):
+            raise ProducerConfigurationError(
+                event_type=event_type,
+                message="Configuration for event_types should be a list or a tuple of dictionaries"
+            )
+        try:
+            signal = OpenEdxPublicSignal.get_signal_by_type(event_type)
+        except KeyError as exc:
+            raise ProducerConfigurationError(message=f"No OpenEdxPublicSignal of type: '{event_type}'.") from exc
+        for configuration in configurations:
+            if not isinstance(configuration, dict):
+                raise ProducerConfigurationError(
+                    event_type=event_type,
+                    message="One of the configuration object is not a dictionary"
+                )
+            expected_keys = {"topic": str, "event_key_field": str, "enabled": bool}
+            for expected_key, expected_type in expected_keys.items():
+                if expected_key not in configuration:
+                    raise ProducerConfigurationError(
+                        event_type=event_type,
+                        message=f"One of the configuration object is missing '{expected_key}' key."
+                    )
+                if not isinstance(configuration[expected_key], expected_type):
+                    raise ProducerConfigurationError(
+                        event_type=event_type,
+                        message=(f"Expected type: {expected_type} for '{expected_key}', "
+                                 f"found: {type(configuration[expected_key])}")
+                    )
+        return signal
+
+    def ready(self):
+        """
+        Read `EVENT_BUS_PRODUCER_CONFIG` setting and connects appropriate handlers to the events based on it.
+
+        Raises:
+            ProducerConfigurationError: If `EVENT_BUS_PRODUCER_CONFIG` is not valid.
+        """
+        load_all_signals()
+        signals_config = getattr(settings, "EVENT_BUS_PRODUCER_CONFIG", {})
+        if not isinstance(signals_config, dict):
+            raise ProducerConfigurationError(
+                message=("Setting 'EVENT_BUS_PRODUCER_CONFIG' should be a dictionary with event_type as"
+                         " key and list or tuple of config dictionaries as values")
+            )
+        for event_type, configurations in signals_config.items():
+            signal = self._get_validated_signal_config(event_type, configurations)
+            signal.connect(general_signal_handler)
+        return super().ready()

openedx_events/exceptions.py

@@ -67,3 +67,23 @@ def __init__(self, event_type="", message=""):
                 event_type=event_type, message=message
             )
         )
+
+
+class ProducerConfigurationError(OpenEdxEventException):
+    """
+    Describes errors that occurs while validating format of producer signal configuration.
+    """
+
+    def __init__(self, event_type="", message=""):
+        """
+        Init method for ProducerConfigurationError custom exception class.
+
+        Arguments:
+            event_type (str): name of the event raising the exception.
+            message (str): message describing why the exception was raised.
+        """
+        super().__init__(
+            message="ProducerConfigurationError {event_type}: {message}".format(
+                event_type=event_type, message=message
+            )
+        )

openedx_events/tests/test_producer_config.py

@@ -0,0 +1,125 @@
+"""
+Test for producer configuration.
+"""
+from unittest.mock import Mock, patch
+
+import ddt
+import pytest
+from django.apps import apps
+from django.test import TestCase, override_settings
+
+from openedx_events.content_authoring.data import XBlockData
+from openedx_events.content_authoring.signals import XBLOCK_DELETED, XBLOCK_PUBLISHED
+from openedx_events.exceptions import ProducerConfigurationError
+
+
+@ddt.ddt
+class ProducerConfiguratonTest(TestCase):
+    """
+    Tests to make sure EVENT_BUS_PRODUCER_CONFIG setting connects required signals to appropriate handlers.
+
+    Attributes:
+        xblock_info: dummy XBlockData.
+    """
+    def setUp(self) -> None:
+        super().setUp()
+        self.xblock_info = XBlockData(
+            usage_key='block-v1:edx+DemoX+Demo_course+type@video+block@UaEBjyMjcLW65gaTXggB93WmvoxGAJa0JeHRrDThk',
+            block_type='video',
+        )
+
+    @patch('openedx_events.apps.get_producer')
+    def test_enabled_disabled_events(self, mock_producer):
+        """
+        Check whether XBLOCK_PUBLISHED is connected to the handler and the handler only publishes enabled events.
+
+        Args:
+            mock_producer: mock get_producer to inspect the arguments.
+        """
+        mock_send = Mock()
+        mock_producer.return_value = mock_send
+        # XBLOCK_PUBLISHED has three configurations where 2 configurations have set enabled as True.
+        XBLOCK_PUBLISHED.send_event(xblock_info=self.xblock_info)
+        mock_send.send.assert_called()
+        mock_send.send.call_count = 2
+
+        # check that call_args_list only consists of enabled topics.
+        call_args = mock_send.send.call_args_list[0][1]
+        self.assertDictContainsSubset(
+            {'topic': 'content-authoring-xblock-lifecycle', 'event_key_field': 'xblock_info.usage_key'},
+            call_args
+        )
+        call_args = mock_send.send.call_args_list[1][1]
+        self.assertDictContainsSubset(
+            {'topic': 'content-authoring-all-status', 'event_key_field': 'xblock_info.usage_key'},
+            call_args
+        )
+
+    @patch('openedx_events.apps.get_producer')
+    @override_settings(EVENT_BUS_PRODUCER_CONFIG={})
+    def test_events_not_in_config(self, mock_producer):
+        """
+        Check whether events not included in the configuration are not published as expected.
+
+        Args:
+            mock_producer: mock get_producer to inspect the arguments.
+        """
+        mock_send = Mock()
+        mock_producer.return_value = mock_send
+        XBLOCK_PUBLISHED.send_event(xblock_info=self.xblock_info)
+        mock_producer.assert_not_called()
+        mock_send.send.assert_not_called()
+
+    def test_configuration_is_validated(self):
+        """
+        Check whether EVENT_BUS_PRODUCER_CONFIG setting is validated before connecting handlers.
+        """
+        with override_settings(EVENT_BUS_PRODUCER_CONFIG=[]):
+            with pytest.raises(ProducerConfigurationError, match="should be a dictionary"):
+                apps.get_app_config("openedx_events").ready()
+
+        with override_settings(EVENT_BUS_PRODUCER_CONFIG={"invalid.event.type": []}):
+            with pytest.raises(ProducerConfigurationError, match="No OpenEdxPublicSignal of type"):
+                apps.get_app_config("openedx_events").ready()
+
+        with override_settings(EVENT_BUS_PRODUCER_CONFIG={"org.openedx.content_authoring.xblock.deleted.v1": ""}):
+            with pytest.raises(ProducerConfigurationError, match="should be a list or a tuple"):
+                apps.get_app_config("openedx_events").ready()
+
+        with override_settings(EVENT_BUS_PRODUCER_CONFIG={"org.openedx.content_authoring.xblock.deleted.v1": [""]}):
+            with pytest.raises(ProducerConfigurationError, match="object is not a dictionary"):
+                apps.get_app_config("openedx_events").ready()
+
+        with override_settings(
+            EVENT_BUS_PRODUCER_CONFIG={
+                "org.openedx.content_authoring.xblock.deleted.v1": [{"topic": "some", "enabled": True}]
+            }
+        ):
+            with pytest.raises(ProducerConfigurationError, match="missing 'event_key_field' key."):
+                apps.get_app_config("openedx_events").ready()
+
+        with override_settings(
+            EVENT_BUS_PRODUCER_CONFIG={
+                "org.openedx.content_authoring.xblock.deleted.v1": [
+                    {"topic": "some", "enabled": 1, "event_key_field": "some"}
+                ]
+            }
+        ):
+            with pytest.raises(
+                ProducerConfigurationError,
+                match="Expected type: <class 'bool'> for 'enabled', found: <class 'int'>"
+            ):
+                apps.get_app_config("openedx_events").ready()
+
+    @patch('openedx_events.apps.get_producer')
+    def test_event_data_key_in_handler(self, mock_producer):
+        """
+        Check whether event_data is constructed properly in handlers.
+        """
+        mock_send = Mock()
+        mock_producer.return_value = mock_send
+        XBLOCK_DELETED.send_event(xblock_info=self.xblock_info)
+        mock_send.send.assert_called_once()
+
+        call_args = mock_send.send.call_args_list[0][1]
+        self.assertIn("xblock_info", call_args["event_data"])

test_utils/test_settings.py

@@ -29,3 +29,29 @@
 )
 
 SECRET_KEY = "not-so-secret-key"
+EVENT_BUS_PRODUCER_CONFIG = {
+    'org.openedx.content_authoring.xblock.published.v1': (
+        {
+            'topic': 'content-authoring-xblock-lifecycle',
+            'event_key_field': 'xblock_info.usage_key',
+            'enabled': True
+        },
+        {
+            'topic': 'content-authoring-all-status',
+            'event_key_field': 'xblock_info.usage_key',
+            'enabled': True
+        },
+        {
+            'topic': 'content-authoring-xblock-published',
+            'event_key_field': 'xblock_info.usage_key',
+            'enabled': False
+        },
+    ),
+    'org.openedx.content_authoring.xblock.deleted.v1': (
+        {
+            'topic': 'content-authoring-xblock-lifecycle',
+            'event_key_field': 'xblock_info.usage_key',
+            'enabled': True
+        },
+    ),
+}