AssignmentsGradesService fix (#46)

* AssignmentsGradesService fix: support pagination, i.e getting data through several requests using "Link" header from the responses

README.rst

@@ -400,6 +400,12 @@ From the service we can get a list of all members by calling:
 
     members = nrps.get_members()
 
+To get some specific page with the members:
+
+.. code-block:: python
+
+    members, next_page_url = nrps.get_members_page(page_url)
+
 Assignments and Grades Service
 ==============================
 
@@ -448,6 +454,24 @@ If you want to send multiple types of grade back, that can be done by specifying
     ags.put_grade(gr, line_item)
 
 If a lineitem with the same ``tag`` exists, that lineitem will be used, otherwise a new lineitem will be created.
+Additional methods:
+
+.. code-block:: python
+
+    # Get one page with line items
+    items_lst, next_page = ags.get_lineitems_page()
+
+    # Get list of all available line items
+    items_lst = ags.get_lineitems()
+
+    # Find line item by ID
+    item = ags.find_lineitem_by_id(ln_id)
+
+    # Find line item by tag
+    item = ags.find_lineitem_by_tag(ln_tag)
+
+    # Return all grades for the passed line item (across all users enrolled in the line item's context)
+    grades = ags.get_grades(ln)
 
 Data privacy launch
 ===================

pylti1p3/__init__.py

@@ -1 +1 @@
-__version__ = '1.9.0'
+__version__ = '1.9.1'

pylti1p3/assignments_grades.py

@@ -27,23 +27,31 @@ def __init__(self, service_connector, service_data):
         self._service_connector = service_connector
         self._service_data = service_data
 
-    def put_grade(self, grade, line_item=None):
+    def put_grade(self, grade, lineitem=None):
         # type: (Grade, t.Optional[LineItem]) -> _ServiceConnectorResponse
+        """
+        Send grade to the LTI platform.
+
+        :param grade: Grade instance
+        :param lineitem: LineItem instance
+        :return: dict with HTTP response body and headers
+        """
+
         if "https://purl.imsglobal.org/spec/lti-ags/scope/score" not in self._service_data['scope']:
             raise LtiException('Missing required scope')
 
-        if line_item and not line_item.get_id():
-            line_item = self.find_or_create_lineitem(line_item)
-            score_url = line_item.get_id()
-        elif not line_item and self._service_data.get('lineitem'):
+        if lineitem and not lineitem.get_id():
+            lineitem = self.find_or_create_lineitem(lineitem)
+            score_url = lineitem.get_id()
+        elif not lineitem and self._service_data.get('lineitem'):
             score_url = self._service_data.get('lineitem')
         else:
-            if not line_item:
-                line_item = LineItem()
-                line_item.set_label('default')\
+            if not lineitem:
+                lineitem = LineItem()
+                lineitem.set_label('default')\
                     .set_score_maximum(100)
-                line_item = self.find_or_create_lineitem(line_item)
-            score_url = line_item.get_id()
+                lineitem = self.find_or_create_lineitem(lineitem)
+            score_url = lineitem.get_id()
 
         assert score_url is not None
         score_url = self._add_url_path_ending(score_url, 'scores')
@@ -55,83 +63,144 @@ def put_grade(self, grade, line_item=None):
             content_type='application/vnd.ims.lis.v1.score+json'
         )
 
-    def get_lineitems(self):
-        # type: () -> list
+    def get_lineitems_page(self, lineitems_url=None):
+        # type: (t.Optional[str]) -> t.Tuple[list, t.Optional[str]]
+        """
+        Get one page with line items.
+
+        :param lineitems_url: LTI platform's URL (optional)
+        :return: tuple in format: (list with line items, next page url)
+        """
         if "https://purl.imsglobal.org/spec/lti-ags/scope/lineitem" not in self._service_data['scope']:
             raise LtiException('Missing required scope')
 
-        line_items = self._service_connector.make_service_request(
+        if not lineitems_url:
+            lineitems_url = self._service_data['lineitems']
+
+        lineitems = self._service_connector.make_service_request(
             self._service_data['scope'],
-            self._service_data['lineitems'],
+            lineitems_url,
             accept='application/vnd.ims.lis.v2.lineitemcontainer+json'
         )
-        if not isinstance(line_items['body'], list):
+        if not isinstance(lineitems['body'], list):
             raise LtiException('Unknown response type received for line items')
-        return line_items['body']
+        return lineitems['body'], lineitems['next_page_url']
+
+    def get_lineitems(self):
+        # type: () -> list
+        """
+        Get list of all available line items.
+
+        :return: list
+        """
+        lineitems_res_lst = []
+        lineitems_url = self._service_data['lineitems']  # type: t.Optional[str]
+
+        while lineitems_url:
+            lineitems, lineitems_url = self.get_lineitems_page(lineitems_url)
+            lineitems_res_lst.extend(lineitems)
+
+        return lineitems_res_lst
+
+    def find_lineitem(self, prop_name, prop_value):
+        # type: (str, t.Any) -> t.Optional[LineItem]
+        """
+        Find line item by some property (ID/Tag).
+
+        :param prop_name: property name
+        :param prop_value: property value
+        :return: LineItem instance or None
+        """
+        lineitems_url = self._service_data['lineitems']  # type: t.Optional[str]
+
+        while lineitems_url:
+            lineitems, lineitems_url = self.get_lineitems_page(lineitems_url)
+            for lineitem in lineitems:
+                lineitem_prop_value = lineitem.get(prop_name)
+                if lineitem_prop_value == prop_value:
+                    return LineItem(lineitem)
+        return None
 
     def find_lineitem_by_id(self, ln_id):
-        # type: (t.Optional[str]) -> t.Optional[LineItem]
-        line_items = self.get_lineitems()
+        # type: (str) -> t.Optional[LineItem]
+        """
+        Find line item by ID.
 
-        for line_item in line_items:
-            line_item_id = line_item.get('id')
-            if line_item_id == ln_id:
-                return LineItem(line_item)
-        return None
+        :param ln_id: str
+        :return: LineItem instance or None
+        """
+        return self.find_lineitem('id', ln_id)
 
     def find_lineitem_by_tag(self, tag):
-        # type: (t.Optional[str]) -> t.Optional[LineItem]
-        line_items = self.get_lineitems()
+        # type: (str) -> t.Optional[LineItem]
+        """
+        Find line item by Tag.
 
-        for line_item in line_items:
-            line_item_tag = line_item.get('tag')
-            if line_item_tag == tag:
-                return LineItem(line_item)
-        return None
+        :param tag: str
+        :return: LineItem instance or None
+        """
+        return self.find_lineitem('tag', tag)
 
-    def find_or_create_lineitem(self, new_line_item, find_by='tag'):
+    def find_or_create_lineitem(self, new_lineitem, find_by='tag'):
         # type: (LineItem, Literal['tag', 'id']) -> LineItem
+        """
+        Try to find line item using ID or Tag. New lime item will be created if nothing is found.
+
+        :param new_lineitem: LineItem instance
+        :param find_by: str ("tag"/"id")
+        :return: LineItem instance (based on response from the LTI platform)
+        """
         if find_by == 'tag':
-            tag = new_line_item.get_tag()
-            line_item = self.find_lineitem_by_tag(tag)
+            tag = new_lineitem.get_tag()
+            if not tag:
+                raise LtiException('Tag value is not specified')
+            lineitem = self.find_lineitem_by_tag(tag)
         elif find_by == 'id':
-            line_id = new_line_item.get_id()
-            line_item = self.find_lineitem_by_id(line_id)
+            line_id = new_lineitem.get_id()
+            if not line_id:
+                raise LtiException('ID value is not specified')
+            lineitem = self.find_lineitem_by_id(line_id)
         else:
             raise LtiException('Invalid "find_by" value: ' + str(find_by))
 
-        if line_item:
-            return line_item
+        if lineitem:
+            return lineitem
 
-        created_line_item = self._service_connector.make_service_request(
+        created_lineitem = self._service_connector.make_service_request(
             self._service_data['scope'],
             self._service_data['lineitems'],
             is_post=True,
-            data=new_line_item.get_value(),
+            data=new_lineitem.get_value(),
             content_type='application/vnd.ims.lis.v2.lineitem+json',
             accept='application/vnd.ims.lis.v2.lineitem+json'
         )
-        if not isinstance(created_line_item['body'], dict):
+        if not isinstance(created_lineitem['body'], dict):
             raise LtiException('Unknown response type received for create line item')
-        return LineItem(created_line_item['body'])
+        return LineItem(created_lineitem['body'])
 
-    def get_grades(self, line_item):
+    def get_grades(self, lineitem):
         # type: (LineItem) -> list
-        line_item_id = line_item.get_id()
-        line_item_tag = line_item.get_tag()
+        """
+        Return all grades for the passed line item (across all users enrolled in the line item's context).
+
+        :param lineitem: LineItem instance
+        :return: list of grades
+        """
+        lineitem_id = lineitem.get_id()
+        lineitem_tag = lineitem.get_tag()
 
         find_by = None  # type: t.Optional[Literal['id', 'tag']]
-        if line_item_id:
+        if lineitem_id:
             find_by = 'id'
-        elif line_item_tag:
+        elif lineitem_tag:
             find_by = 'tag'
         else:
             raise LtiException('Received LineItem did not contain a tag or id')
 
-        line_item = self.find_or_create_lineitem(line_item, find_by=find_by)
-        line_item_id = line_item.get_id()
-        assert line_item_id is not None
-        results_url = self._add_url_path_ending(line_item_id, 'results')
+        lineitem = self.find_or_create_lineitem(lineitem, find_by=find_by)
+        lineitem_id = lineitem.get_id()
+        assert lineitem_id is not None
+        results_url = self._add_url_path_ending(lineitem_id, 'results')
         scores = self._service_connector.make_service_request(
             self._service_data['scope'],
             results_url,

pylti1p3/names_roles.py

@@ -1,4 +1,3 @@
-import re
 import typing as t
 
 if t.TYPE_CHECKING:
@@ -34,26 +33,37 @@ def __init__(self, service_connector, service_data):
         self._service_connector = service_connector
         self._service_data = service_data
 
+    def get_members_page(self, members_url=None):
+        # type: (t.Optional[str]) -> t.Tuple[list, t.Optional[str]]
+        """
+        Get one page with the users.
+
+        :param members_url: LTI platform's URL (optional)
+        :return: tuple in format: (list with users, next page url)
+        """
+        if not members_url:
+            members_url = self._service_data['context_memberships_url']
+
+        data = self._service_connector.make_service_request(
+            ['https://purl.imsglobal.org/spec/lti-nrps/scope/contextmembership.readonly'],
+            members_url,
+            accept='application/vnd.ims.lti-nrps.v2.membershipcontainer+json',
+        )
+        data_body = t.cast(t.Any, data.get('body', {}))
+        return data_body.get('members', []), data['next_page_url']
+
     def get_members(self):
         # type: () -> t.List[_Member]
-        members = []  # type: t.List[_Member]
-        next_page = self._service_data['context_memberships_url']  # type: t.Union[Literal[False], str]
-
-        while next_page:
-            page = self._service_connector.make_service_request(
-                ['https://purl.imsglobal.org/spec/lti-nrps/scope/contextmembership.readonly'],
-                next_page,  # type: ignore
-                accept='application/vnd.ims.lti-nrps.v2.membershipcontainer+json',
-                case_insensitive_headers=True
-            )
-
-            members.extend(t.cast(t.Any, page.get('body', {})).get('members', []))
-
-            next_page = False
-            link_header = page.get('headers', {}).get('link', '')
-            if link_header:
-                match = re.search(r'<([^>]*)>;\s*rel="next"', link_header.replace('\n', ' ').lower().strip())
-                if match:
-                    next_page = match.group(1)
-
-        return members
+        """
+        Get list with all users.
+
+        :return: list
+        """
+        members_res_lst = []  # type: t.List[_Member]
+        members_url = self._service_data['context_memberships_url']  # type: t.Optional[str]
+
+        while members_url:
+            members, members_url = self.get_members_page(members_url)
+            members_res_lst.extend(members)
+
+        return members_res_lst

pylti1p3/service_connector.py

@@ -1,4 +1,5 @@
 import hashlib
+import re
 import sys
 import time
 import typing as t
@@ -16,6 +17,7 @@
     _ServiceConnectorResponse = TypedDict('_ServiceConnectorResponse', {
         'headers': t.Union[t.Dict[str, str], t.MutableMapping[str, str]],
         'body': t.Union[None, int, float, t.List[object], t.Dict[str, object], str],
+        'next_page_url': t.Optional[str]
     })
 
 
@@ -118,7 +120,15 @@ def make_service_request(
         if not r.ok:
             raise LtiServiceException(r)
 
+        next_page_url = None
+        link_header = r.headers.get('link', '')
+        if link_header:
+            match = re.search(r'<([^>]*)>;\s*rel="next"', link_header.replace('\n', ' ').lower().strip())
+            if match:
+                next_page_url = match.group(1)
+
         return {
             'headers': r.headers if case_insensitive_headers else dict(r.headers),
-            'body': r.json() if r.content else None
+            'body': r.json() if r.content else None,
+            'next_page_url': next_page_url if next_page_url else None
         }