[BD-32] feat: add async feature to Hooks Extension Framework tooling

edx_django_utils/hooks/pipeline.py → edx_django_utils/hooks/tasks.py

@@ -3,12 +3,15 @@
 """
 from logging import getLogger
 
+from celery import shared_task
+
 from .exceptions import HookException
 from .utils import get_functions_for_pipeline
 
 log = getLogger(__name__)
 
 
+@shared_task()
 def run_pipeline(pipeline, *args, raise_exception=False, **kwargs):
     """
     Given a list of functions paths, this function  will execute them using the Accumulative Pipeline

edx_django_utils/hooks/tests/test_pipeline.py

@@ -6,7 +6,7 @@
 from django.test import TestCase
 
 from ..exceptions import HookException
-from ..pipeline import run_pipeline
+from ..tasks import run_pipeline
 
 
 class TestRunningPipeline(TestCase):
@@ -24,7 +24,7 @@ def setUp(self):
         }
         self.pipeline = Mock()
 
-    @patch("edx_django_utils.hooks.pipeline.get_functions_for_pipeline")
+    @patch("edx_django_utils.hooks.tasks.get_functions_for_pipeline")
     def test_run_empty_pipeline(self, get_functions_mock):
         """
         This method runs an empty pipeline, i.e, a pipeline without defined functions.
@@ -39,7 +39,7 @@ def test_run_empty_pipeline(self, get_functions_mock):
         get_functions_mock.assert_called_once_with([])
         self.assertEqual(result, self.kwargs)
 
-    @patch("edx_django_utils.hooks.pipeline.get_functions_for_pipeline")
+    @patch("edx_django_utils.hooks.tasks.get_functions_for_pipeline")
     def test_raise_hook_exception(self, get_functions_mock):
         """
         This method runs a pipeline with a function that raises HookException.
@@ -61,7 +61,7 @@ def test_raise_hook_exception(self, get_functions_mock):
             captured.records[0].getMessage(), log_message,
         )
 
-    @patch("edx_django_utils.hooks.pipeline.get_functions_for_pipeline")
+    @patch("edx_django_utils.hooks.tasks.get_functions_for_pipeline")
     def test_not_raise_hook_exception(self, get_functions_mock):
         """
         This method runs a pipeline with a function that raises HookException but
@@ -85,7 +85,7 @@ def test_not_raise_hook_exception(self, get_functions_mock):
         self.assertEqual(result, return_value)
         function_without_exception.assert_called_once_with(**self.kwargs)
 
-    @patch("edx_django_utils.hooks.pipeline.get_functions_for_pipeline")
+    @patch("edx_django_utils.hooks.tasks.get_functions_for_pipeline")
     def test_not_raise_common_exception(self, get_functions_mock):
         """
         This method runs a pipeline with a function that raises a common Exception.
@@ -117,7 +117,7 @@ def test_not_raise_common_exception(self, get_functions_mock):
         self.assertEqual(result, return_value)
         function_without_exception.assert_called_once_with(**self.kwargs)
 
-    @patch("edx_django_utils.hooks.pipeline.get_functions_for_pipeline")
+    @patch("edx_django_utils.hooks.tasks.get_functions_for_pipeline")
     def test_getting_pipeline_result(self, get_functions_mock):
         """
         This method runs a pipeline with functions defined via configuration.
@@ -145,7 +145,7 @@ def test_getting_pipeline_result(self, get_functions_mock):
         second_function.assert_called_once_with(**return_value_1st)
         self.assertDictEqual(result, return_overall_value)
 
-    @patch("edx_django_utils.hooks.pipeline.get_functions_for_pipeline")
+    @patch("edx_django_utils.hooks.tasks.get_functions_for_pipeline")
     def test_partial_pipeline(self, get_functions_mock):
         """
         This method runs a pipeline with functions defined via configuration.

edx_django_utils/hooks/tests/test_triggers.py

@@ -99,7 +99,6 @@ def test_async_filter_pipeline(self, get_configuration_mock, run_pipeline_mock):
 
         Expected behavior:
             A task is started using the pipeline function.
-        TODO: add mock to async call when the feature is added.
         """
         pipeline = Mock()
         is_async = True
@@ -110,7 +109,7 @@ def test_async_filter_pipeline(self, get_configuration_mock, run_pipeline_mock):
 
         trigger_filter("trigger_name", **self.kwargs)
 
-        run_pipeline_mock.assert_called_once_with(
+        run_pipeline_mock.delay.assert_called_once_with(
             pipeline, raise_exception=True, **self.kwargs
         )
 
@@ -197,7 +196,6 @@ def test_async_action_pipeline(self, get_configuration_mock, run_pipeline_mock):
 
         Expected behavior:
             A task is started using the pipeline function.
-        TODO: add mock to async call when the feature is added.
         """
         pipeline, is_async = Mock(), True
         get_configuration_mock.return_value = (
@@ -207,4 +205,4 @@ def test_async_action_pipeline(self, get_configuration_mock, run_pipeline_mock):
 
         trigger_action("trigger_name", **self.kwargs)
 
-        run_pipeline_mock.assert_called_once_with(pipeline, **self.kwargs)
+        run_pipeline_mock.delay.assert_called_once_with(pipeline, **self.kwargs)

edx_django_utils/hooks/triggers.py

@@ -1,9 +1,15 @@
 """
 Triggers for actions and filters.
 """
-from .pipeline import run_pipeline
+from logging import getLogger
+
+from kombu.exceptions import EncodeError
+
+from .tasks import run_pipeline
 from .utils import get_pipeline_configuration
 
+log = getLogger(__name__)
+
 
 def trigger_filter(trigger_name, *args, **kwargs):
     """
@@ -35,10 +41,17 @@ def trigger_filter(trigger_name, *args, **kwargs):
     if not pipeline:
         return kwargs
 
+    result = kwargs
+
     if is_async:
-        result = run_pipeline(
-            pipeline, *args, raise_exception=True, **kwargs
-        )  # TODO: change to async call.
+        try:
+            result = run_pipeline.delay(pipeline, *args, raise_exception=True, **kwargs)
+        except (TypeError, EncodeError):
+            log.exception(
+                "An error ocurred in trigger_filter while executing `run pipeline` with arguments: %s, %s.",
+                str(args),
+                str(kwargs),
+            )
     else:
         result = run_pipeline(pipeline, *args, raise_exception=True, **kwargs)
 
@@ -70,6 +83,13 @@ def trigger_action(trigger_name, *args, **kwargs):
         return
 
     if is_async:
-        run_pipeline(pipeline, *args, **kwargs)  # TODO: change to async call.
+        try:
+            run_pipeline.delay(pipeline, *args, **kwargs)
+        except (TypeError, EncodeError):
+            log.exception(
+                "An error ocurred in trigger_action while executing `run_pipeline` with arguments: %s, %s.",
+                str(args),
+                str(kwargs),
+            )
     else:
         run_pipeline(pipeline, *args, **kwargs)

requirements/base.in

@@ -4,6 +4,7 @@
 
 Django                    # Web application framework
 django-waffle             # Allows for feature toggles in Django.
+celery                    # Allows async execution.
 newrelic                  # New Relic agent for performance monitoring
 psutil                    # Library for retrieving information on running processes and system utilization
                           # NOTE: psutil pinned to match edx-platform version. Will need to be updated at the same time.

requirements/base.txt

@@ -4,6 +4,14 @@
 #
 #    make upgrade
 #
+amqp==2.6.1
+    # via kombu
+billiard==3.6.3.0
+    # via celery
+celery==4.4.7
+    # via
+    #   -c requirements/constraints.txt
+    #   -r requirements/base.in
 django-crum==0.7.9
     # via -r requirements/base.in
 django-waffle==2.1.0
@@ -13,15 +21,23 @@ django==2.2.19
     #   -c https://raw.githubusercontent.com/edx/edx-lint/master/edx_lint/files/common_constraints.txt
     #   -r requirements/base.in
     #   django-crum
+kombu==4.6.11
+    # via celery
 newrelic==6.2.0.156
     # via -r requirements/base.in
 pbr==5.5.1
     # via stevedore
 psutil==5.8.0
     # via -r requirements/base.in
 pytz==2021.1
-    # via django
+    # via
+    #   celery
+    #   django
 sqlparse==0.4.1
     # via django
 stevedore==3.3.0
     # via -r requirements/base.in
+vine==1.3.0
+    # via
+    #   amqp
+    #   celery

requirements/constraints.txt

@@ -10,5 +10,6 @@
 
 # Common constraints for edx repos
 -c https://raw.githubusercontent.com/edx/edx-lint/master/edx_lint/files/common_constraints.txt
-
 
+# Constraint to use same version as in edx-platform.
+celery<5.0

requirements/dev.txt

@@ -4,6 +4,10 @@
 #
 #    make upgrade
 #
+amqp==2.6.1
+    # via
+    #   -r requirements/quality.txt
+    #   kombu
 appdirs==1.4.4
     # via
     #   -r requirements/ci.txt
@@ -17,6 +21,14 @@ attrs==20.3.0
     # via
     #   -r requirements/quality.txt
     #   pytest
+billiard==3.6.3.0
+    # via
+    #   -r requirements/quality.txt
+    #   celery
+celery==4.4.7
+    # via
+    #   -c requirements/constraints.txt
+    #   -r requirements/quality.txt
 certifi==2020.12.5
     # via
     #   -r requirements/ci.txt
@@ -101,6 +113,10 @@ jinja2==2.11.3
     #   code-annotations
     #   diff-cover
     #   jinja2-pluralize
+kombu==4.6.11
+    # via
+    #   -r requirements/quality.txt
+    #   celery
 lazy-object-proxy==1.6.0
     # via
     #   -r requirements/quality.txt
@@ -203,6 +219,7 @@ python-slugify==4.0.1
 pytz==2021.1
     # via
     #   -r requirements/quality.txt
+    #   celery
     #   django
 pyyaml==5.4.1
     # via
@@ -256,6 +273,11 @@ urllib3==1.26.4
     # via
     #   -r requirements/ci.txt
     #   requests
+vine==1.3.0
+    # via
+    #   -r requirements/quality.txt
+    #   amqp
+    #   celery
 virtualenv==20.4.3
     # via
     #   -r requirements/ci.txt

requirements/doc.txt

@@ -6,14 +6,26 @@
 #
 alabaster==0.7.12
     # via sphinx
+amqp==2.6.1
+    # via
+    #   -r requirements/test.txt
+    #   kombu
 attrs==20.3.0
     # via
     #   -r requirements/test.txt
     #   pytest
 babel==2.9.0
     # via sphinx
+billiard==3.6.3.0
+    # via
+    #   -r requirements/test.txt
+    #   celery
 bleach==3.3.0
     # via readme-renderer
+celery==4.4.7
+    # via
+    #   -c requirements/constraints.txt
+    #   -r requirements/test.txt
 certifi==2020.12.5
     # via requests
 cffi==1.14.5
@@ -28,7 +40,7 @@ coverage==5.5
     # via
     #   -r requirements/test.txt
     #   pytest-cov
-cryptography==3.4.6
+cryptography==3.4.7
     # via secretstorage
 ddt==1.4.2
     # via -r requirements/test.txt
@@ -69,8 +81,12 @@ jeepney==0.6.0
     #   secretstorage
 jinja2==2.11.3
     # via sphinx
-keyring==23.0.0
+keyring==23.0.1
     # via twine
+kombu==4.6.11
+    # via
+    #   -r requirements/test.txt
+    #   celery
 markupsafe==1.1.1
     # via jinja2
 mock==4.0.3
@@ -123,6 +139,7 @@ pytz==2021.1
     # via
     #   -r requirements/test.txt
     #   babel
+    #   celery
     #   django
 readme-renderer==29.0
     # via
@@ -183,6 +200,11 @@ twine==3.4.1
     # via -r requirements/doc.in
 urllib3==1.26.4
     # via requests
+vine==1.3.0
+    # via
+    #   -r requirements/test.txt
+    #   amqp
+    #   celery
 webencodings==0.5.1
     # via bleach
 zipp==3.4.1