refactor: removed linting errors

openedx_cmi5_xblock/openedx_cmi5_xblock.py

@@ -40,6 +40,7 @@
 
 CMI5XML_FILENAME = 'cmi5.xml'
 
+
 def _(text):
     return text
 
@@ -173,14 +174,20 @@ def student_view(self, context=None):
     def lrs_endpoint(self, request, _suffix):
         """
         Handles requests related to the Learning Record Store (LRS) endpoint.
-        
+
         Also sends the xapi statements obtained to the external LRS
         """
-        lrs_url = "https://cloud.scorm.com/lrs/M4W0NJJ0HY/statements/"
+        credentials = self.get_credentials()
 
         if request.params.get('statementId') and request.method == 'PUT':
             statement_data = get_request_body(request)
-            send_xapi_to_external_lrs(statement_data, lrs_url)
+            # send statements to external lrs.
+            send_xapi_to_external_lrs(
+                statement_data,
+                credentials["EXTERNAL_LRS_URL"],
+                credentials["ACTIVITY_PROVIDER_KEY"],
+                credentials["SECRET_KEY"]
+                )
 
             lesson_status = statement_data.get('verb').get('display').get('en')
             object_categories = statement_data.get('context', {}).get('contextActivities', {}).get('category')
@@ -199,17 +206,16 @@ def lrs_endpoint(self, request, _suffix):
 
         elif request.params.get('stateId'):
             state_id = request.params.get('stateId')
-            
+
             if state_id == 'LMS.LaunchData':
                 return Response(json.dumps(self.get_launch_state_data()), status=200)
             elif state_id == 'suspendData' and request.method == 'GET':
                 return json_response(self.state_data)
             elif state_id == 'suspendData' and request.method == 'PUT':
                 self.state_data = get_request_body(request)
                 return Response(status=204)
-        
+
         return json_response({'success': True})
-
 
     @XBlock.handler
     def lrs_auth_endpoint(self, request, _suffix):
@@ -250,7 +256,30 @@ def studio_submit(self, request, _suffix):
             response['errors'].append(e.args[0])
         return json_response(response)
 
-    #getters and setters
+    # getters and setters
+    def get_credentials(self):
+        """
+        Retrieves credentials from XBlock settings.
+
+        If any of the credentials is not found in the XBlock settings, an empty string is provided as a default value.
+        """
+        EXTERNAL_LRS_URL = self.xblock_settings.get(
+            "EX_LRS_ENDPOINT", ""
+        )
+
+        ACTIVITY_PROVIDER_KEY = self.xblock_settings.get(
+            "ACTIVITY_PROVIDER_KEY", ""
+        )
+        SECRET_KEY = self.xblock_settings.get(
+            "SECRET_KEY", ""
+        )
+
+        return {
+            "EXTERNAL_LRS_URL": EXTERNAL_LRS_URL,
+            "ACTIVITY_PROVIDER_KEY": ACTIVITY_PROVIDER_KEY,
+            "SECRET_KEY": SECRET_KEY
+        }
+
     def get_current_user_attr(self, attr: str):
         """Get the value of a specific attribute for the current user."""
         return self.get_current_user().opt_attrs.get(attr)

openedx_cmi5_xblock/utils/utility.py

@@ -7,8 +7,8 @@
 from django.core.validators import URLValidator
 from webob import Response
 
-USERNAME = 'uEy9xZTOzB3fEr_JY78'
-PASSWORD = 's4qMwv9fDFceov_JxOU'
+# USERNAME = 'uEy9xZTOzB3fEr_JY78'
+# PASSWORD = 's4qMwv9fDFceov_JxOU'
 
 
 def json_response(data):
@@ -57,33 +57,39 @@ def get_sha1(file_descriptor):
     return sha1.hexdigest()
 
 
-def send_xapi_to_external_lrs(xapi_data, lrs_url):
+def send_xapi_to_external_lrs(xapi_data, lrs_url, ACTIVITY_PROVIDER_KEY, SECRET_KEY):
     """Send xAPI data to the specified LRS URL."""
-
+    timeout = 10
     headers = {
-    'Content-Type': 'application/json',
-    'X-Experience-API-Version': '1.0.3'
+        'Content-Type': 'application/json',
+        'X-Experience-API-Version': '1.0.3'
     }
 
     try:
-        response = requests.post(lrs_url, headers=headers, auth=HTTPBasicAuth(USERNAME, PASSWORD), data=json.dumps(xapi_data))
+        response = requests.post(
+            lrs_url,
+            headers=headers,
+            auth=HTTPBasicAuth(ACTIVITY_PROVIDER_KEY, SECRET_KEY),
+            data=json.dumps(xapi_data),
+            timeout=timeout
+        )
         response.raise_for_status()
-        
+
         print("Successfully sent xAPI data to LRS.")
         print(f"Response Status Code: {response.status_code}")
         print(f"Response Content: {response.text}")
 
     except requests.exceptions.HTTPError as errh:
-        print ("HTTP Error:",errh)
+        print("HTTP Error:", errh)
 
     except requests.exceptions.ConnectionError as errc:
-        print ("Error Connecting:",errc)
+        print("Error Connecting:", errc)
 
     except requests.exceptions.Timeout as errt:
-        print ("Timeout Error:",errt)
+        print("Timeout Error:", errt)
 
     except requests.exceptions.RequestException as err:
-        print ("Error:",err)
+        print("Error:", err)
 
 
 def parse_int(value, default):