Address Issue #74 - [DOC] add typings to functions (#81)

* [DOC] add typings - replace iterritems - add unit tests

ChangeLog.md

@@ -7,8 +7,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
-## Fixed
-- Fixed quickstart tutorial when installed via pip
+## [v0.3.9] - 2024-08-29
+### Added
+- Added type hints
+- Added more unit tests for daatapi.py
+
+### Fixed
+- fixed usage of deprecated iterritems to items
 
 ## [v0.3.8] - 2024-05-28
 ### Added

learnosity_sdk/_version.py

@@ -1 +1 @@
-__version__ = 'v0.3.8'
+__version__ = 'v0.3.9'

learnosity_sdk/request/dataapi.py

@@ -1,3 +1,5 @@
+from typing import Any, Dict, Generator
+from requests import Response
 import requests
 import copy
 
@@ -7,8 +9,8 @@
 
 class DataApi(object):
 
-    def request(self, endpoint, security_packet,
-                secret, request_packet={}, action='get'):
+    def request(self, endpoint: str, security_packet: Dict[str, str],
+                secret: str, request_packet:Dict[str, Any] = {}, action: str = 'get') -> Response:
         """
         Make a request to Data API
 
@@ -33,9 +35,9 @@ def request(self, endpoint, security_packet,
         init = Init('data', security_packet, secret, request_packet, action)
         return requests.post(endpoint, data=init.generate())
 
-    def results_iter(self, endpoint, security_packet,
-                     secret, request_packet={},
-                     action='get'):
+    def results_iter(self, endpoint: str, security_packet: Dict[str, str],
+                     secret: str, request_packet: Dict[str, Any] = {},
+                     action:str = 'get') -> Generator[Dict[str, Any], None, None]:
         """
         Return an iterator of all results from a request to Data API
 
@@ -60,15 +62,15 @@ def results_iter(self, endpoint, security_packet,
                                           secret, request_packet,
                                           action):
             if type(response['data']) == dict:
-                for key, value in response['data'].iteritems():
+                for key, value in response['data'].items():
                     yield {key: value}
             else:
                 for result in response['data']:
                     yield result
 
-    def request_iter(self, endpoint, security_packet,
-                     secret, request_packet={},
-                     action='get'):
+    def request_iter(self, endpoint: str, security_packet: Dict[str, str],
+                     secret: str, request_packet: Dict[str, Any] = {},
+                     action: str = 'get') -> Generator[Dict[str, Any], None, None]:
         """
         Iterate over the pages of results of a query to data api
 

learnosity_sdk/request/init.py

@@ -4,12 +4,13 @@
 import hmac
 import json
 import platform
+from typing import Any, Dict, Iterable, Optional, Union
 from learnosity_sdk._version import __version__
 
 from learnosity_sdk.exceptions import ValidationException
 
 
-def format_utc_time():
+def format_utc_time() -> str:
     "Get the current UTC time, formatted for a security timestamp"
     now = datetime.datetime.utcnow()
     return now.strftime("%Y%m%d-%H%M")
@@ -33,8 +34,9 @@ class Init(object):
     __telemetry_enabled = True
 
     def __init__(
-            self, service, security, secret,
-            request=None, action=None):
+            self, service: str, security: Dict[str, Any], secret: str,
+            request: Optional[Dict[str, Any]] = None, action:Optional[str] = None) -> None:
+        # Using None as a default value will throw mypy typecheck issues. This should be addressed
         self.service = service
         self.security = security.copy()
         self.secret = secret
@@ -50,10 +52,10 @@ def __init__(
         self.set_service_options()
         self.security['signature'] = self.generate_signature()
 
-    def is_telemetry_enabled(self):
+    def is_telemetry_enabled(self) -> bool:
         return self.__telemetry_enabled
 
-    def generate(self, encode=True):
+    def generate(self, encode: bool = True) -> Union[str, Dict[str, Any]]:
         """
         Generate the data necessary to make a request to one of the Learnosity
         products/services.
@@ -106,7 +108,7 @@ def generate(self, encode=True):
         else:
             return output
 
-    def get_sdk_meta(self):
+    def get_sdk_meta(self) -> Dict[str, str]:
         return {
             'version': self.get_sdk_version(),
             'lang': 'python',
@@ -115,15 +117,15 @@ def get_sdk_meta(self):
             'platform_version': platform.release()
         }
 
-    def get_sdk_version(self):
+    def get_sdk_version(self) -> str:
         return __version__
 
-    def generate_request_string(self):
+    def generate_request_string(self) -> Union[str, None]:
         if self.request is None:
             return None
         return json.dumps(self.request, separators=(',', ':'), ensure_ascii=False)
 
-    def generate_signature(self):
+    def generate_signature(self) -> str:
 
         vals = []
 
@@ -142,7 +144,7 @@ def generate_signature(self):
 
         return self.hash_list(vals)
 
-    def validate(self):
+    def validate(self) -> None:
         # Parse the security packet if the user provided it as a string
         if isinstance(self.security, str):
             self.security = json.loads(self.security)
@@ -185,7 +187,7 @@ def validate(self):
                 'user_id' not in self.security:
             raise ValidationException("questions API requires a user id")
 
-    def set_service_options(self):
+    def set_service_options(self) -> None:
         if self.service == 'questions':
             self.sign_request_data = False
         elif self.service == 'assess':
@@ -235,13 +237,13 @@ def set_service_options(self):
             if len(hashed_users) > 0:
                 self.security['users'] = hashed_users
 
-    def hash_list(self, l):
+    def hash_list(self, l: Iterable[Any]) -> str:
         "Hash a list by concatenating values with an underscore"
         concatValues = "_".join(l)
         signature =  hmac.new(bytes(str(self.secret),'utf_8'), msg = bytes(str(concatValues) , 'utf-8'), digestmod = hashlib.sha256).hexdigest()
         return '$02$' + signature
 
-    def add_telemetry_data(self):
+    def add_telemetry_data(self) -> None:
         if self.__telemetry_enabled:
             if 'meta' in self.request:
                 self.request['meta']['sdk'] = self.get_sdk_meta()
@@ -256,9 +258,9 @@ def add_telemetry_data(self):
     """
 
     @classmethod
-    def disable_telemetry(cls):
+    def disable_telemetry(cls) -> None:
         cls.__telemetry_enabled = False
 
     @classmethod
-    def enable_telemetry(cls):
+    def enable_telemetry(cls) -> None:
         cls.__telemetry_enabled = True

learnosity_sdk/utils/lrnuuid.py

@@ -2,5 +2,5 @@
 
 class Uuid:
     @staticmethod
-    def generate():
+    def generate() -> str:
         return str(uuid.uuid4())

tests/unit/test_dataapi.py

@@ -1,68 +1,70 @@
 import unittest
 import responses
 from learnosity_sdk.request import DataApi
-
-# This test uses the consumer key and secret for the demos consumer
-# this is the only consumer with publicly available keys
-security = {
-    'consumer_key': 'yis0TYCu7U9V4o7M',
-    'domain': 'demos.learnosity.com'
-}
-# WARNING: Normally the consumer secret should not be committed to a public
-# repository like this one. Only this specific key is publically available.
-consumer_secret = '74c5fd430cf1242a527f6223aebd42d30464be22'
-request = {
-    # These items should already exist for the demos consumer
-    'references': ['item_2', 'item_3'],
-    'limit': 1
-}
-action = 'get'
-endpoint = 'https://data.learnosity.com/v1/itembank/items'
-dummy_responses = [{
-    'meta': {
-        'status': True,
-        'timestamp': 1514874527,
-        'records': 2,
-        'next': '1'
-    },
-    'data': [{'id': 'a'}]
-}, {
-    'meta': {
-        'status': True,
-        'timestamp': 1514874527,
-        'records': 2
-    },
-    'data': [{'id': 'b'}]
-}]
-
+from learnosity_sdk.exceptions import DataApiException
 
 class UnitTestDataApiClient(unittest.TestCase):
     """
     Tests to ensure that the Data API client functions correctly.
     """
 
+    def setUp(self):
+        # This test uses the consumer key and secret for the demos consumer
+        # this is the only consumer with publicly available keys
+        self.security = {
+            'consumer_key': 'yis0TYCu7U9V4o7M',
+            'domain': 'demos.learnosity.com'
+        }
+        # WARNING: Normally the consumer secret should not be committed to a public
+        # repository like this one. Only this specific key is publically available.
+        self.consumer_secret = '74c5fd430cf1242a527f6223aebd42d30464be22'
+        self.request = {
+            # These items should already exist for the demos consumer
+            'references': ['item_2', 'item_3'],
+            'limit': 1
+        }
+        self.action = 'get'
+        self.endpoint = 'https://data.learnosity.com/v1/itembank/items'
+        self.dummy_responses = [{
+            'meta': {
+                'status': True,
+                'timestamp': 1514874527,
+                'records': 2,
+                'next': '1'
+            },
+            'data': [{'id': 'a'}]
+        }, {
+            'meta': {
+                'status': True,
+                'timestamp': 1514874527,
+                'records': 2
+            },
+            'data': [{'id': 'b'}]
+        }]
+        self.invalid_json = "This is not valid JSON!"
+
     @responses.activate
     def test_request(self):
         """
         Verify that `request` sends a request after it has been signed
         """
-        for dummy in dummy_responses:
-            responses.add(responses.POST, endpoint, json=dummy)
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json=dummy)
         client = DataApi()
-        res = client.request(endpoint, security, consumer_secret, request,
-                             action)
-        assert res.json() == dummy_responses[0]
-        assert responses.calls[0].request.url == endpoint
+        res = client.request(self.endpoint, self.security, self.consumer_secret, self.request,
+                             self.action)
+        assert res.json() == self.dummy_responses[0]
+        assert responses.calls[0].request.url == self.endpoint
         assert 'signature' in responses.calls[0].request.body
 
     @responses.activate
     def test_request_iter(self):
         """Verify that `request_iter` returns an iterator of pages"""
-        for dummy in dummy_responses:
-            responses.add(responses.POST, endpoint, json=dummy)
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json=dummy)
         client = DataApi()
-        pages = client.request_iter(endpoint, security, consumer_secret,
-                                    request, action)
+        pages = client.request_iter(self.endpoint, self.security, self.consumer_secret,
+                                    self.request, self.action)
         results = []
         for page in pages:
             results.append(page)
@@ -74,13 +76,47 @@ def test_request_iter(self):
     @responses.activate
     def test_results_iter(self):
         """Verify that `result_iter` returns an iterator of results"""
-        for dummy in dummy_responses:
-            responses.add(responses.POST, endpoint, json=dummy)
+        self.dummy_responses[1]['data'] = {'id': 'b'}
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json=dummy)
         client = DataApi()
-        result_iter = client.results_iter(endpoint, security, consumer_secret,
-                                          request, action)
+        result_iter = client.results_iter(self.endpoint, self.security, self.consumer_secret,
+                                          self.request, self.action)
         results = list(result_iter)
 
         assert len(results) == 2
         assert results[0]['id'] == 'a'
         assert results[1]['id'] == 'b'
+
+    @responses.activate
+    def test_results_iter_error_status(self):
+        """Verify that a DataApiException is raised http status is not ok"""
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json={}, status=500)
+        client = DataApi()
+        with self.assertRaisesRegex(DataApiException, "server returned HTTP status 500"):
+            list(client.results_iter(self.endpoint, self.security, self.consumer_secret,
+                                     self.request, self.action))
+
+    @responses.activate
+    def test_results_iter_no_meta_status(self):
+        """Verify that a DataApiException is raised when 'meta' 'status' is None"""
+        for response in self.dummy_responses:
+            response['meta']['status'] = None
+
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json=dummy)
+        client = DataApi()
+        with self.assertRaisesRegex(DataApiException, "server returned unsuccessful status:"):
+            list(client.results_iter(self.endpoint, self.security, self.consumer_secret,
+                                     self.request, self.action))
+
+    @responses.activate
+    def test_results_iter_invalid_response_data(self):
+        """Verify that a DataApiException is raised response data isn't valid JSON"""
+        for dummy in self.dummy_responses:
+            responses.add(responses.POST, self.endpoint, json=None)
+        client = DataApi()
+        with self.assertRaisesRegex(DataApiException, "server returned invalid json: "):
+            list(client.results_iter(self.endpoint, self.security, self.consumer_secret,
+                                     self.request, self.action))