feat: utility to reset application state before processing event

chore: bump version and update changelog

chore: remove utility function from coverage

docs: move how to add new implementation doc from atlassian

docs: simplify consuming docs in how-to

CHANGELOG.rst

@@ -14,6 +14,12 @@ Change Log
 Unreleased
 ----------
 
+[9.3.0] - 2024-01-17
+--------------------
+Added
+~~~~~
+* Adds utility function to reset application state similar to setup/teardown in Django request/response cycle.
+
 [9.2.0] - 2023-11-16
 --------------------
 Added

docs/how-tos/add-new-event-bus-concrete-implementation.rst

@@ -0,0 +1,35 @@
+How to add a new concrete implementation of the event bus
+=========================================================
+
+Context
+-------
+
+Here is a list of the existing concrete implementations of the event bus:
+
+- `Kafka <https://github.com/openedx/event-bus-kafka>`_
+- `Redis Streams <https://github.com/openedx/event-bus-redis>`_
+
+This how-to is to help you add a new concrete implementation, for example using Pulsar or some other technology.
+
+Producing
+---------
+
+There should be a producer class that inherits from `EventBusProducer <https://github.com/openedx/openedx-events/blob/cbb59f124ed84afacb9ec99baa82a86381370dcc/openedx_events/event_bus/__init__.py#L66>`_ in openedx-events.
+
+The defined ``send`` method is meant to be called from within a signal receiver in the producing service.
+
+Consuming
+---------
+
+At a high level, the consumer should be a process that takes the signals and events from the broker and emits the signal with the event. There should be a consumer class that inherits from `EventBusConsumer <https://github.com/openedx/openedx-events/blob/06635f3642cee4020d6787df68bba694bd1233fe/openedx_events/event_bus/__init__.py#L127>`_ in openedx-events.
+
+The consumer class then needs to implement ``consume_indefinitely`` loop, which will stay running and listen to events as they come in.
+
+We have included an utility function called `prepare_for_new_work_cycle <../../openedx_events/tooling.py#L323>`_ in openedx-events which needs to be called before processing any signal. Currently, it reconnects the db connection if required as well as clears RequestCache and there may be later, more comprehensive changes. These steps mimic some setup/teardown that is normally performed by Django in its request/response based architecture.
+
+Checkout `consumer.py <https://github.com/openedx/event-bus-redis/blob/main/edx_event_bus_redis/internal/consumer.py>`_ in event bus redis implementation.
+
+Abstraction tickets
+-------------------
+
+The known remaining work for a fully abstracted event bus is captured in the `Abstraction tickets <https://github.com/orgs/edx/projects/11/views/4?filterQuery=label%3Aevent-bus+-status%3ADone+abstraction>`_

docs/how-tos/index.rst

@@ -9,3 +9,4 @@ How-tos
    adding-events-to-a-service
    adding-events-to-event-bus
    using-events
+   add-new-event-bus-concrete-implementation

openedx_events/__init__.py

@@ -5,4 +5,4 @@
 more information about the project.
 """
 
-__version__ = "9.2.0"
+__version__ = "9.3.0"

openedx_events/tooling.py

@@ -7,7 +7,9 @@
 from logging import getLogger
 
 from django.conf import settings
+from django.db import connection
 from django.dispatch import Signal
+from edx_django_utils.cache import RequestCache
 
 from openedx_events.data import EventsMetadata
 from openedx_events.exceptions import SenderValidationError
@@ -292,3 +294,45 @@ def load_all_signals():
     Loads all non-test signals.py modules.
     """
     _process_all_signals_modules(import_module)
+
+
+def _reconnect_to_db_if_needed():  # pragma: no cover
+    """
+    Reconnects the db connection if needed.
+
+    This is important because Django only does connection validity/age checks as part of
+    its request/response cycle, which isn't in effect for the consume-loop. If we don't
+    force these checks, a broken connection will remain broken indefinitely. For most
+    consumers, this will cause event processing to fail.
+    """
+    has_connection = bool(connection.connection)
+    requires_reconnect = has_connection and not connection.is_usable()
+    if requires_reconnect:
+        connection.connect()
+
+
+def _clear_request_cache():  # pragma: no cover
+    """
+    Clear the RequestCache so that each event consumption starts fresh.
+
+    Signal handlers may be written with the assumption that they are called in the context
+    of a web request, so we clear the request cache just in case.
+    """
+    RequestCache.clear_all_namespaces()
+
+
+def prepare_for_new_work_cycle():  # pragma: no cover
+    """
+    Ensure that the application state is appropriate for performing a new unit of work.
+
+    This mimics some setup/teardown that is normally performed by Django in its
+    request/response based architecture and that is needed for ensuring a clean and
+    usable state in this worker-based application.
+
+    See https://github.com/openedx/openedx-events/issues/236 for details.
+    """
+    # Ensure that the database connection is active and usable.
+    _reconnect_to_db_if_needed()
+
+    # Clear the request cache, in case anything in the signal handlers rely on it.
+    _clear_request_cache()

requirements/base.in

@@ -1,6 +1,7 @@
 # Core requirements for using this application
 -c constraints.txt
 
+edx_django_utils
 django
 attrs
 fastavro

requirements/base.txt

@@ -8,24 +8,47 @@ asgiref==3.7.2
     # via django
 attrs==23.1.0
     # via -r requirements/base.in
-django==3.2.21
+cffi==1.16.0
+    # via pynacl
+click==8.1.7
+    # via edx-django-utils
+django==3.2.22
     # via
     #   -c requirements/common_constraints.txt
     #   -r requirements/base.in
+    #   django-crum
+    #   django-waffle
+    #   edx-django-utils
+django-crum==0.7.9
+    # via edx-django-utils
+django-waffle==4.0.0
+    # via edx-django-utils
+edx-django-utils==5.7.0
+    # via -r requirements/base.in
 edx-opaque-keys[django]==2.5.1
     # via -r requirements/base.in
-fastavro==1.8.3
+fastavro==1.8.4
     # via -r requirements/base.in
+newrelic==9.1.0
+    # via edx-django-utils
 pbr==5.11.1
     # via stevedore
+psutil==5.9.5
+    # via edx-django-utils
+pycparser==2.21
+    # via cffi
 pymongo==3.13.0
     # via edx-opaque-keys
+pynacl==1.5.0
+    # via edx-django-utils
 pytz==2023.3.post1
     # via django
 sqlparse==0.4.4
     # via django
 stevedore==5.1.0
-    # via edx-opaque-keys
+    # via
+    #   edx-django-utils
+    #   edx-opaque-keys
 typing-extensions==4.8.0
     # via
     #   asgiref

requirements/ci.txt

@@ -12,7 +12,7 @@ filelock==3.12.4
     #   virtualenv
 packaging==23.2
     # via tox
-platformdirs==3.10.0
+platformdirs==3.11.0
     # via virtualenv
 pluggy==1.3.0
     # via tox

requirements/dev.txt

@@ -27,6 +27,7 @@ cffi==1.16.0
     # via
     #   -r requirements/quality.txt
     #   cryptography
+    #   pynacl
 chardet==5.2.0
     # via diff-cover
 charset-normalizer==3.3.0
@@ -39,6 +40,7 @@ click==8.1.7
     #   -r requirements/quality.txt
     #   click-log
     #   code-annotations
+    #   edx-django-utils
     #   edx-lint
     #   pip-tools
 click-log==0.4.0
@@ -49,7 +51,7 @@ code-annotations==1.5.0
     # via
     #   -r requirements/quality.txt
     #   edx-lint
-coverage[toml]==7.3.1
+coverage[toml]==7.3.2
     # via
     #   -r requirements/quality.txt
     #   pytest-cov
@@ -69,14 +71,27 @@ distlib==0.3.7
     # via
     #   -r requirements/ci.txt
     #   virtualenv
-django==3.2.21
+django==3.2.22
     # via
     #   -c requirements/common_constraints.txt
     #   -r requirements/quality.txt
+    #   django-crum
+    #   django-waffle
+    #   edx-django-utils
+django-crum==0.7.9
+    # via
+    #   -r requirements/quality.txt
+    #   edx-django-utils
+django-waffle==4.0.0
+    # via
+    #   -r requirements/quality.txt
+    #   edx-django-utils
 docutils==0.20.1
     # via
     #   -r requirements/quality.txt
     #   readme-renderer
+edx-django-utils==5.7.0
+    # via -r requirements/quality.txt
 edx-lint==5.3.4
     # via -r requirements/quality.txt
 edx-opaque-keys[django]==2.5.1
@@ -85,7 +100,7 @@ exceptiongroup==1.1.3
     # via
     #   -r requirements/quality.txt
     #   pytest
-fastavro==1.8.3
+fastavro==1.8.4
     # via -r requirements/quality.txt
 filelock==3.12.4
     # via
@@ -157,6 +172,10 @@ more-itertools==10.1.0
     # via
     #   -r requirements/quality.txt
     #   jaraco-classes
+newrelic==9.1.0
+    # via
+    #   -r requirements/quality.txt
+    #   edx-django-utils
 nh3==0.2.14
     # via
     #   -r requirements/quality.txt
@@ -179,7 +198,7 @@ pkginfo==1.9.6
     # via
     #   -r requirements/quality.txt
     #   twine
-platformdirs==3.10.0
+platformdirs==3.11.0
     # via
     #   -r requirements/ci.txt
     #   -r requirements/quality.txt
@@ -192,11 +211,15 @@ pluggy==1.3.0
     #   diff-cover
     #   pytest
     #   tox
+psutil==5.9.5
+    # via
+    #   -r requirements/quality.txt
+    #   edx-django-utils
 py==1.11.0
     # via
     #   -r requirements/ci.txt
     #   tox
-pycodestyle==2.11.0
+pycodestyle==2.11.1
     # via -r requirements/quality.txt
 pycparser==2.21
     # via
@@ -234,6 +257,10 @@ pymongo==3.13.0
     # via
     #   -r requirements/quality.txt
     #   edx-opaque-keys
+pynacl==1.5.0
+    # via
+    #   -r requirements/quality.txt
+    #   edx-django-utils
 pyproject-hooks==1.0.0
     # via
     #   -r requirements/pip-tools.txt
@@ -302,6 +329,7 @@ stevedore==5.1.0
     # via
     #   -r requirements/quality.txt
     #   code-annotations
+    #   edx-django-utils
     #   edx-opaque-keys
 text-unidecode==1.3
     # via
@@ -337,7 +365,7 @@ typing-extensions==4.8.0
     #   edx-opaque-keys
     #   pylint
     #   rich
-urllib3==2.0.5
+urllib3==2.0.6
     # via
     #   -r requirements/quality.txt
     #   requests