diff --git a/README.rst b/README.rst index a16e57d..b751570 100644 --- a/README.rst +++ b/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 =================== diff --git a/pylti1p3/__init__.py b/pylti1p3/__init__.py index e5102d3..35424e8 100644 --- a/pylti1p3/__init__.py +++ b/pylti1p3/__init__.py @@ -1 +1 @@ -__version__ = '1.9.0' +__version__ = '1.9.1' diff --git a/pylti1p3/assignments_grades.py b/pylti1p3/assignments_grades.py index 8bc9de6..2ef3600 100644 --- a/pylti1p3/assignments_grades.py +++ b/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, diff --git a/pylti1p3/names_roles.py b/pylti1p3/names_roles.py index 6aee27a..13a25dd 100644 --- a/pylti1p3/names_roles.py +++ b/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 diff --git a/pylti1p3/service_connector.py b/pylti1p3/service_connector.py index 77faf6b..5a652e8 100644 --- a/pylti1p3/service_connector.py +++ b/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 }