feat: trigger grade change event to message bus

lms/envs/common.py

@@ -4851,3 +4851,6 @@ def _make_locale_paths(settings):  # pylint: disable=missing-function-docstring
 # The expected value is an Integer representing the cutoff point (in months) for inclusion to the message. Example:
 # a value of `3` would include learners who have logged in within the past 3 months.
 BULK_COURSE_EMAIL_LAST_LOGIN_ELIGIBILITY_PERIOD = None
+
+################# Kafka ################
+KAFKA_ENABLED = False

lms/envs/devstack.py

@@ -161,7 +161,6 @@ def should_show_debug_toolbar(request):  # lint-amnesty, pylint: disable=missing
 ################################ COURSE LICENSES ################################
 FEATURES['LICENSING'] = True
 
-
 ########################## Courseware Search #######################
 FEATURES['ENABLE_COURSEWARE_SEARCH'] = False
 FEATURES['ENABLE_COURSEWARE_SEARCH_FOR_COURSE_STAFF'] = True
@@ -438,6 +437,18 @@ def should_show_debug_toolbar(request):  # lint-amnesty, pylint: disable=missing
 #################### Webpack Configuration Settings ##############################
 WEBPACK_LOADER['DEFAULT']['TIMEOUT'] = 5
 
+#################### Kafka Related Settings ##############################
+KAFKA_PRODUCER_CONF_BASE = {'bootstrap.servers': "edx.devstack.kafka:29092",
+                 'client.id': 'edx.devstack.lms' }
+
+KAFKA_CONSUMER_CONF_BASE = {'bootstrap.servers': "edx.devstack.kafka:29092",
+                       'group.id': 'lms'}
+
+
+SCHEMA_REGISTRY_URL = "http://edx.devstack.schema-registry:8081"
+
+KAFKA_ENABLED = True
+
 ################# New settings must go ABOVE this line #################
 ########################################################################
 # See if the developer has any local overrides.

openedx/core/djangoapps/credentials/grade_change_event.py

@@ -0,0 +1,68 @@
+from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
+from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka import SerializingProducer
+from confluent_kafka.serialization import StringSerializer
+from uuid import uuid4
+from django.conf import settings
+
+
+grade_change_schema_string = """
+    {
+        "namespace": "djangoapps.credentials.producer",
+        "name": "GradeChangeEvent",
+        "type": "record",
+        "fields": [
+            {"name": "username", "type": "string"},
+            {"name":"course_run", "type": "string"},
+            {"name":"letter_grade", "type": ["string", "null"], "default": "null"},
+            {"name":"percent_grade", "type": ["float", "null"], "default": "null"},
+            {"name":"verified", "type": "boolean"}
+        ]
+    }
+"""
+
+class GradeChangeEvent:
+    def __init__(self, *args, **kwargs):
+        self.course_run = kwargs['course_run']
+        self.username = kwargs['username']
+        self.letter_grade = kwargs['letter_grade']
+        self.percent_grade = kwargs['percent_grade']
+        self.verified = kwargs['verified']
+
+    @staticmethod
+    def from_dict(obj, ctx):
+        return GradeChangeEvent(obj)
+
+    @staticmethod
+    def to_dict(grade_change_event, ctx):
+        return {
+            "course_run": grade_change_event.course_run,
+            "username": grade_change_event.username,
+            "letter_grade": grade_change_event.letter_grade,
+            "percent_grade": grade_change_event.percent_grade,
+            "verified": grade_change_event.verified,
+        }
+
+
+def produce_grade_change_event(user, course_run_key, letter_grade, percent_grade, verified):
+    schema_registry_client = SchemaRegistryClient({
+        'url': settings.SCHEMA_REGISTRY_URL,
+    })
+    grade_change_event_serializer = AvroSerializer(schema_str=grade_change_schema_string,
+                                                   schema_registry_client=schema_registry_client,
+                                                   to_dict=GradeChangeEvent.to_dict)
+
+    producer_settings = dict(settings.KAFKA_PRODUCER_CONF_BASE)
+    producer_settings.update({'key.serializer': StringSerializer('utf-8'),
+                              'value.serializer': grade_change_event_serializer})
+    grade_change_event_producer = SerializingProducer(producer_settings)
+    grade_change_event_producer.produce("credentials_grade_change", key=str(uuid4()),
+                                        value=GradeChangeEvent(
+                                            username=getattr(user, 'username', None),
+                                            course_run=str(course_run_key),
+                                            letter_grade=letter_grade,
+                                            percent_grade=percent_grade,
+                                            verified=verified
+                                        ))
+    grade_change_event_producer.poll()
+

openedx/core/djangoapps/credentials/tasks/v1/tasks.py

@@ -15,6 +15,7 @@
 from edx_django_utils.monitoring import set_code_owner_attribute
 from MySQLdb import OperationalError
 from opaque_keys.edx.keys import CourseKey
+from uuid import uuid4
 
 from common.djangoapps.course_modes.models import CourseMode
 from lms.djangoapps.certificates.api import get_recently_modified_certificates
@@ -28,6 +29,7 @@
 from openedx.core.djangoapps.credentials.utils import get_credentials_api_client
 from openedx.core.djangoapps.programs.signals import handle_course_cert_awarded, handle_course_cert_changed
 from openedx.core.djangoapps.site_configuration.models import SiteConfiguration
+from openedx.core.djangoapps.credentials.grade_change_event import produce_grade_change_event
 
 logger = get_task_logger(__name__)
 
@@ -279,6 +281,7 @@ def send_grade_if_interesting(user, course_run_key, mode, status, letter_grade,
                 verbose=verbose
             )
         logger.info(msg)
+
     # Avoid scheduling new tasks if certification is disabled. (Grades are a part of the records/cert story)
     if not CredentialsApiConfig.current().is_learner_issuance_enabled:
         if verbose:
@@ -349,6 +352,8 @@ def send_grade_if_interesting(user, course_run_key, mode, status, letter_grade,
         letter_grade = grade.letter_grade
         percent_grade = grade.percent
 
+    if settings.KAFKA_ENABLED:
+        produce_grade_change_event(user, str(course_run_key), letter_grade, percent_grade, True)
     send_grade_to_credentials.delay(user.username, str(course_run_key), True, letter_grade, percent_grade)
 
 

requirements/common_constraints.txt

@@ -3,6 +3,11 @@
 # See BOM-2721 for more details.
 # Below is the copied and edited version of common_constraints
 
+# This is a temporary solution to override the real common_constraints.txt
+# In edx-lint, until the pyjwt constraint in edx-lint has been removed.
+# See BOM-2721 for more details.
+# Below is the copied and edited version of common_constraints
+
 # A central location for most common version constraints
 # (across edx repos) for pip-installation.
 #

requirements/edx/base.in

@@ -164,3 +164,5 @@ XBlock                              # Courseware component architecture
 xblock-utils                        # Provides utilities used by the Discussion XBlock
 xss-utils                           # https://github.com/edx/edx-platform/pull/20633 Fix XSS via Translations
 enmerkar-underscore                 # Implements a underscore extractor for django-babel.
+confluent-kafka                     # Client for working with kafka used by the arch_experiments djangoapp.
+fastavro==1.4.7

requirements/edx/base.txt

@@ -129,6 +129,8 @@ code-annotations==1.2.0
     # via
     #   edx-enterprise
     #   edx-toggles
+confluent-kafka==1.7.0
+    # via -r requirements/edx/base.in
 contextlib2==21.6.0
     # via -r requirements/edx/base.in
 coreapi==2.3.3
@@ -519,6 +521,8 @@ event-tracking==1.1.4
     #   -r requirements/edx/base.in
     #   edx-proctoring
     #   edx-search
+fastavro==1.4.7
+    # via -r requirements/edx/base.in
 frozenlist==1.2.0
     # via
     #   aiohttp

requirements/edx/development.txt

@@ -188,6 +188,8 @@ code-annotations==1.2.0
     #   edx-enterprise
     #   edx-lint
     #   edx-toggles
+confluent-kafka==1.7.0
+    # via -r requirements/edx/testing.txt
 contextlib2==21.6.0
     # via -r requirements/edx/testing.txt
 coreapi==2.3.3
@@ -652,6 +654,8 @@ fastapi==0.70.0
     # via
     #   -r requirements/edx/testing.txt
     #   pact-python
+fastavro==1.4.7
+    # via -r requirements/edx/testing.txt
 filelock==3.3.2
     # via
     #   -r requirements/edx/testing.txt

requirements/edx/testing.txt

@@ -77,7 +77,6 @@ attrs==21.2.0
     #   aiohttp
     #   edx-ace
     #   openedx-events
-    #   outcome
     #   pytest
 babel==2.9.1
     # via
@@ -177,6 +176,8 @@ code-annotations==1.2.0
     #   edx-enterprise
     #   edx-lint
     #   edx-toggles
+confluent-kafka==1.7.0
+    # via -r requirements/edx/base.txt
 contextlib2==21.6.0
     # via -r requirements/edx/base.txt
 coreapi==2.3.3
@@ -627,6 +628,8 @@ faker==9.8.2
     # via factory-boy
 fastapi==0.70.0
     # via pact-python
+fastavro==1.4.7
+    # via -r requirements/edx/base.txt
 filelock==3.3.2
     # via
     #   tox