lms.model.backend

   1import logging
   2import typing
   3
   4import edq.util.parse
   5
   6import lms.model.assignments
   7import lms.model.constants
   8import lms.model.courses
   9import lms.model.groups
  10import lms.model.groupsets
  11import lms.model.query
  12import lms.model.scores
  13import lms.model.users
  14
  15_logger = logging.getLogger(__name__)
  16
  17class APIBackend():
  18    """
  19    API backends provide a unified interface to an LMS.
  20
  21    Note that instead of using an abstract class,
  22    methods will raise a NotImplementedError by default.
  23    This will allow child backends to fill in as much functionality as they can,
  24    while still leaving gaps where they are incomplete or impossible.
  25    """
  26
  27    _testing_override: typing.Union[bool, None] = None
  28    """ A top-level override to control testing status. """
  29
  30    def __init__(self,
  31            server: str,
  32            backend_type: str,
  33            testing: typing.Union[bool, str] = False,
  34            **kwargs: typing.Any) -> None:
  35        self.server: str = server
  36        """ The server this backend will connect to. """
  37
  38        self.backend_type: str = backend_type
  39        """
  40        The type for this backend.
  41        Should be set by the child class.
  42        """
  43
  44        parsed_testing = edq.util.parse.boolean(testing)
  45        if (APIBackend._testing_override is not None):
  46            parsed_testing = APIBackend._testing_override
  47
  48        self.testing: bool = parsed_testing
  49        """ True if the backend is being used for a test. """
  50
  51    # Core Methods
  52
  53    def is_testing(self) -> bool:
  54        """ Check if this backend is in testing mode. """
  55
  56        return self.testing
  57
  58    def get_standard_headers(self) -> typing.Dict[str, str]:
  59        """
  60        Get standard headers for this backend.
  61        Children should take care to set the write header when performing a write operation.
  62        """
  63
  64        return {
  65            lms.model.constants.HEADER_KEY_BACKEND: self.backend_type,
  66            lms.model.constants.HEADER_KEY_WRITE: 'false',
  67        }
  68
  69    def not_found(self, operation: str, identifiers: typing.Dict[str, typing.Any]) -> None:
  70        """
  71        Called when the backend was unable to find some object.
  72        This will only be called when a requested object is not found,
  73        e.g., a user requested by ID is not found.
  74        This is not called when a list naturally returns zero results,
  75        or when a query does not match any items.
  76        """
  77
  78        _logger.warning("Object not found during operation: '%s'. Identifiers: %s.", operation, identifiers)
  79
  80    # API Methods
  81
  82    def courses_get(self,
  83            course_queries: typing.Collection[lms.model.courses.CourseQuery],
  84            **kwargs: typing.Any) -> typing.List[lms.model.courses.Course]:
  85        """
  86        Get the specified courses associated with the given course.
  87        """
  88
  89        if (len(course_queries) == 0):
  90            return []
  91
  92        courses = self.courses_list(**kwargs)
  93
  94        matches = []
  95        for course in sorted(courses):
  96            for query in course_queries:
  97                if (query.match(course)):
  98                    matches.append(course)
  99                    break
 100
 101        return sorted(matches)
 102
 103    def courses_fetch(self,
 104            course_id: str,
 105            **kwargs: typing.Any) -> typing.Union[lms.model.courses.Course, None]:
 106        """
 107        Fetch a single course associated with the context user.
 108        Return None if no matching course is found.
 109
 110        By default, this will just do a list and choose the relevant record.
 111        Specific backends may override this if there are performance concerns.
 112        """
 113
 114        courses = self.courses_list(**kwargs)
 115        for course in courses:
 116            if (course.id == course_id):
 117                return course
 118
 119        return None
 120
 121    def courses_list(self,
 122            **kwargs: typing.Any) -> typing.List[lms.model.courses.Course]:
 123        """
 124        List the courses associated with the context user.
 125        """
 126
 127        raise NotImplementedError('courses_list')
 128
 129    def courses_assignments_get(self,
 130            course_query: lms.model.courses.CourseQuery,
 131            assignment_queries: typing.Collection[lms.model.assignments.AssignmentQuery],
 132            **kwargs: typing.Any) -> typing.List[lms.model.assignments.Assignment]:
 133        """
 134        Get the specified assignments associated with the given course.
 135        """
 136
 137        if (len(assignment_queries) == 0):
 138            return []
 139
 140        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 141
 142        assignments = sorted(self.courses_assignments_list(resolved_course_query.get_id(), **kwargs))
 143        assignment_queries = sorted(assignment_queries)
 144
 145        matches = []
 146        for assignment in assignments:
 147            for query in assignment_queries:
 148                if (query.match(assignment)):
 149                    matches.append(assignment)
 150                    break
 151
 152        return matches
 153
 154    def courses_assignments_fetch(self,
 155            course_id: str,
 156            assignment_id: str,
 157            **kwargs: typing.Any) -> typing.Union[lms.model.assignments.Assignment, None]:
 158        """
 159        Fetch a single assignment associated with the given course.
 160        Return None if no matching assignment is found.
 161
 162        By default, this will just do a list and choose the relevant record.
 163        Specific backends may override this if there are performance concerns.
 164        """
 165
 166        assignments = self.courses_assignments_list(course_id, **kwargs)
 167        for assignment in sorted(assignments):
 168            if (assignment.id == assignment_id):
 169                return assignment
 170
 171        return None
 172
 173    def courses_assignments_list(self,
 174            course_id: str,
 175            **kwargs: typing.Any) -> typing.List[lms.model.assignments.Assignment]:
 176        """
 177        List the assignments associated with the given course.
 178        """
 179
 180        raise NotImplementedError('courses_assignments_list')
 181
 182    def courses_assignments_resolve_and_list(self,
 183            course_query: lms.model.courses.CourseQuery,
 184            **kwargs: typing.Any) -> typing.List[lms.model.assignments.Assignment]:
 185        """
 186        List the assignments associated with the given course.
 187        """
 188
 189        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 190        return sorted(self.courses_assignments_list(resolved_course_query.get_id(), **kwargs))
 191
 192    def courses_assignments_scores_get(self,
 193            course_query: lms.model.courses.CourseQuery,
 194            assignment_query: lms.model.assignments.AssignmentQuery,
 195            user_queries: typing.Collection[lms.model.users.UserQuery],
 196            **kwargs: typing.Any) -> typing.List[lms.model.scores.AssignmentScore]:
 197        """
 198        Get the scores associated with the given assignment query and user queries.
 199        """
 200
 201        if (len(user_queries) == 0):
 202            return []
 203
 204        scores = self.courses_assignments_scores_resolve_and_list(course_query, assignment_query, **kwargs)
 205
 206        matches = []
 207        for score in scores:
 208            for user_query in user_queries:
 209                if (user_query.match(score.user)):
 210                    matches.append(score)
 211
 212        return sorted(matches)
 213
 214    def courses_assignments_scores_fetch(self,
 215            course_id: str,
 216            assignment_id: str,
 217            user_id: str,
 218            **kwargs: typing.Any) -> typing.Union[lms.model.scores.AssignmentScore, None]:
 219        """
 220        Fetch the score associated with the given assignment and user.
 221
 222        By default, this will just do a list and choose the relevant record.
 223        Specific backends may override this if there are performance concerns.
 224        """
 225
 226        scores = self.courses_assignments_scores_list(course_id, assignment_id, **kwargs)
 227        for score in scores:
 228            if ((score.user is not None) and (score.user.id == user_id)):
 229                return score
 230
 231        return None
 232
 233    def courses_assignments_scores_list(self,
 234            course_id: str,
 235            assignment_id: str,
 236            **kwargs: typing.Any) -> typing.List[lms.model.scores.AssignmentScore]:
 237        """
 238        List the scores associated with the given assignment.
 239        """
 240
 241        raise NotImplementedError('courses_assignments_scores_list')
 242
 243    def courses_assignments_scores_resolve_and_list(self,
 244            course_query: lms.model.courses.CourseQuery,
 245            assignment_query: lms.model.assignments.AssignmentQuery,
 246            **kwargs: typing.Any) -> typing.List[lms.model.scores.AssignmentScore]:
 247        """
 248        List the scores associated with the given assignment query.
 249        In addition to resolving the assignment query,
 250        users will also be resolved into their full version
 251        (instead of the reduced version usually returned with scores).
 252        """
 253
 254        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 255
 256        # Resolve the assignment query.
 257        matched_assignments = self.courses_assignments_get(resolved_course_query, [assignment_query], **kwargs)
 258        if (len(matched_assignments) == 0):
 259            return []
 260
 261        target_assignment = matched_assignments[0]
 262
 263        # List the scores.
 264        scores = self.courses_assignments_scores_list(resolved_course_query.get_id(), target_assignment.id, **kwargs)
 265        if (len(scores) == 0):
 266            return []
 267
 268        # Resolve the scores' queries.
 269
 270        users = self.courses_users_list(resolved_course_query.get_id(), **kwargs)
 271        users_map = {user.id: user for user in users}
 272
 273        for score in scores:
 274            score.assignment = target_assignment.to_query()
 275
 276            if ((score.user is not None) and (score.user.id in users_map)):
 277                score.user = users_map[score.user.id].to_query()
 278
 279        return sorted(scores)
 280
 281    def courses_assignments_scores_resolve_and_upload(self,
 282            course_query: lms.model.courses.CourseQuery,
 283            assignment_query: lms.model.assignments.AssignmentQuery,
 284            scores: typing.Dict[lms.model.users.UserQuery, lms.model.scores.ScoreFragment],
 285            **kwargs: typing.Any) -> int:
 286        """
 287        Resolve queries and upload assignment scores (indexed by user query).
 288        A None score (ScoreFragment.score) indicates that the score should be cleared.
 289        Return the number of scores sent to the LMS.
 290        """
 291
 292        if (len(scores) == 0):
 293            return 0
 294
 295        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 296        resolved_assignment_query = self.resolve_assignment_query(resolved_course_query.get_id(), assignment_query, **kwargs)
 297
 298        resolved_users = self.resolve_user_queries(resolved_course_query.get_id(), list(scores.keys()), warn_on_miss = True)
 299        resolved_scores: typing.Dict[str, lms.model.scores.ScoreFragment] = {}
 300
 301        for (user, score) in scores.items():
 302            for resolved_user in resolved_users:
 303                if (user.match(resolved_user)):
 304                    resolved_scores[resolved_user.get_id()] = score
 305                    continue
 306
 307        if (len(resolved_scores) == 0):
 308            return 0
 309
 310        return self.courses_assignments_scores_upload(
 311                resolved_course_query.get_id(),
 312                resolved_assignment_query.get_id(),
 313                resolved_scores,
 314                **kwargs)
 315
 316    def courses_assignments_scores_upload(self,
 317            course_id: str,
 318            assignment_id: str,
 319            scores: typing.Dict[str, lms.model.scores.ScoreFragment],
 320            **kwargs: typing.Any) -> int:
 321        """
 322        Upload assignment scores (indexed by user id).
 323        A None score (ScoreFragment.score) indicates that the score should be cleared.
 324        Return the number of scores sent to the LMS.
 325        """
 326
 327        raise NotImplementedError('courses_assignments_scores_upload')
 328
 329    def courses_gradebook_get(self,
 330            course_query: lms.model.courses.CourseQuery,
 331            assignment_queries: typing.Collection[lms.model.assignments.AssignmentQuery],
 332            user_queries: typing.Collection[lms.model.users.UserQuery],
 333            **kwargs: typing.Any) -> lms.model.scores.Gradebook:
 334        """
 335        Get a gradebook with the specified users and assignments.
 336        Specifying no users/assignments is the same as requesting all of them.
 337        """
 338
 339        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 340
 341        resolved_assignment_queries = self.resolve_assignment_queries(resolved_course_query.get_id(), assignment_queries, empty_all = True, **kwargs)
 342        assignment_ids = [query.get_id() for query in resolved_assignment_queries]
 343
 344        resolved_user_queries = self.resolve_user_queries(resolved_course_query.get_id(), user_queries,
 345                empty_all = True, only_students = True, **kwargs)
 346        user_ids = [query.get_id() for query in resolved_user_queries]
 347
 348        gradebook = self.courses_gradebook_fetch(resolved_course_query.get_id(), assignment_ids, user_ids, **kwargs)
 349
 350        # Resolve the gradebook's queries (so it can show names/emails instead of just IDs).
 351        gradebook.update_queries(resolved_assignment_queries, resolved_user_queries)
 352
 353        return gradebook
 354
 355    def courses_gradebook_fetch(self,
 356            course_id: str,
 357            assignment_ids: typing.Collection[str],
 358            user_ids: typing.Collection[str],
 359            **kwargs: typing.Any) -> lms.model.scores.Gradebook:
 360        """
 361        Get a gradebook with the specified users and assignments.
 362        If either the assignments or users is empty, an empty gradebook will be returned.
 363        """
 364
 365        raise NotImplementedError('courses_gradebook_fetch')
 366
 367    def courses_gradebook_list(self,
 368            course_id: str,
 369            **kwargs: typing.Any) -> lms.model.scores.Gradebook:
 370        """
 371        List the full gradebook associated with this course.
 372        """
 373
 374        return self.courses_gradebook_get(lms.model.courses.CourseQuery(id = course_id), [], [], **kwargs)
 375
 376    def courses_gradebook_resolve_and_list(self,
 377            course_query: lms.model.courses.CourseQuery,
 378            **kwargs: typing.Any) -> lms.model.scores.Gradebook:
 379        """
 380        List the full gradebook associated with this course.
 381        """
 382
 383        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 384        return self.courses_gradebook_list(resolved_course_query.get_id(), **kwargs)
 385
 386    def courses_gradebook_resolve_and_upload(self,
 387            course_query: lms.model.courses.CourseQuery,
 388            gradebook: lms.model.scores.Gradebook,
 389            **kwargs: typing.Any) -> int:
 390        """
 391        Resolve queries and upload a gradebook.
 392        Missing scores in the gradebook are skipped,
 393        a None score (ScoreFragment.score) indicates that the score should be cleared.
 394        Return the number of scores sent to the LMS.
 395        """
 396
 397        if (len(gradebook) == 0):
 398            return 0
 399
 400        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 401
 402        assignments = self.courses_assignments_list(resolved_course_query.get_id(), **kwargs)
 403        users = self.courses_users_list(resolved_course_query.get_id(), **kwargs)
 404
 405        resolved_assignment_queries = [assignment.to_query() for assignment in assignments]
 406        resolved_user_queries = [user.to_query() for user in users]
 407
 408        gradebook.update_queries(resolved_assignment_queries, resolved_user_queries)
 409
 410        return self.courses_gradebook_upload(
 411                resolved_course_query.get_id(),
 412                gradebook,
 413                **kwargs)
 414
 415    def courses_gradebook_upload(self,
 416            course_id: str,
 417            gradebook: lms.model.scores.Gradebook,
 418            **kwargs: typing.Any) -> int:
 419        """
 420        Upload a gradebook.
 421        All queries in the gradebook must be resolved (or at least have an ID).
 422        Missing scores in the gradebook are skipped,
 423        a None score (ScoreFragment.score) indicates that the score should be cleared.
 424        Return the number of scores sent to the LMS.
 425        """
 426
 427        assignment_scores = gradebook.get_scores_by_assignment()
 428
 429        count = 0
 430        for (assignment, user_scores) in assignment_scores.items():
 431            if (assignment.id is None):
 432                raise ValueError(f"Assignment query for gradebook upload ({assignment}) does not have an ID.")
 433
 434            upload_scores = {}
 435            for (user, score) in user_scores.items():
 436                if (user.id is None):
 437                    raise ValueError(f"User query for gradebook upload ({user}) does not have an ID.")
 438
 439                upload_scores[user.id] = score.to_fragment()
 440
 441            count += self.courses_assignments_scores_upload(course_id, assignment.id, upload_scores, **kwargs)
 442
 443        return count
 444
 445    def courses_groupsets_create(self,
 446            course_id: str,
 447            name: str,
 448            **kwargs: typing.Any) -> lms.model.groupsets.GroupSet:
 449        """
 450        Create a group set.
 451        """
 452
 453        raise NotImplementedError('courses_groupsets_create')
 454
 455    def courses_groupsets_resolve_and_create(self,
 456            course_query: lms.model.courses.CourseQuery,
 457            name: str,
 458            **kwargs: typing.Any) -> lms.model.groupsets.GroupSet:
 459        """
 460        Resolve references and create a group set.
 461        """
 462
 463        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 464        return self.courses_groupsets_create(resolved_course_query.get_id(), name, **kwargs)
 465
 466    def courses_groupsets_delete(self,
 467            course_id: str,
 468            groupset_id: str,
 469            **kwargs: typing.Any) -> bool:
 470        """
 471        Delete a group set.
 472        """
 473
 474        raise NotImplementedError('courses_groupsets_delete')
 475
 476    def courses_groupsets_resolve_and_delete(self,
 477            course_query: lms.model.courses.CourseQuery,
 478            groupset_query: lms.model.groupsets.GroupSetQuery,
 479            **kwargs: typing.Any) -> bool:
 480        """
 481        Resolve references and create a group set.
 482        """
 483
 484        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 485        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 486        return self.courses_groupsets_delete(resolved_course_query.get_id(), resolved_groupset_query.get_id(), **kwargs)
 487
 488    def courses_groupsets_get(self,
 489            course_query: lms.model.courses.CourseQuery,
 490            groupset_queries: typing.Collection[lms.model.groupsets.GroupSetQuery],
 491            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSet]:
 492        """
 493        Get the specified group sets associated with the given course.
 494        """
 495
 496        if (len(groupset_queries) == 0):
 497            return []
 498
 499        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 500        groupset_queries = sorted(groupset_queries)
 501        groupsets = sorted(self.courses_groupsets_list(resolved_course_query.get_id(), **kwargs))
 502
 503        matches = []
 504        for groupset in groupsets:
 505            for query in groupset_queries:
 506                if (query.match(groupset)):
 507                    matches.append(groupset)
 508                    break
 509
 510        return matches
 511
 512    def courses_groupsets_fetch(self,
 513            course_id: str,
 514            groupset_id: str,
 515            **kwargs: typing.Any) -> typing.Union[lms.model.groupsets.GroupSet, None]:
 516        """
 517        Fetch a single group set associated with the given course.
 518        Return None if no matching group set is found.
 519
 520        By default, this will just do a list and choose the relevant record.
 521        Specific backends may override this if there are performance concerns.
 522        """
 523
 524        groupsets = self.courses_groupsets_list(course_id, **kwargs)
 525        for groupset in groupsets:
 526            if (groupset.id == groupset_id):
 527                return groupset
 528
 529        return None
 530
 531    def courses_groupsets_list(self,
 532            course_id: str,
 533            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSet]:
 534        """
 535        List the group sets associated with the given course.
 536        """
 537
 538        raise NotImplementedError('courses_groupsets_list')
 539
 540    def courses_groupsets_resolve_and_list(self,
 541            course_query: lms.model.courses.CourseQuery,
 542            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSet]:
 543        """
 544        List the group sets associated with the given course.
 545        """
 546
 547        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 548        return sorted(self.courses_groupsets_list(resolved_course_query.get_id(), **kwargs))
 549
 550    def courses_groupsets_memberships_resolve_and_add(self,
 551            course_query: lms.model.courses.CourseQuery,
 552            groupset_query: lms.model.groupsets.GroupSetQuery,
 553            memberships: typing.Collection[lms.model.groups.GroupMembership],
 554            **kwargs: typing.Any) -> typing.Tuple[
 555                    typing.List[lms.model.groups.Group],
 556                    typing.Dict[lms.model.groups.ResolvedGroupQuery, int]
 557            ]:
 558        """
 559        Resolve queries and add the specified users to the specified groups.
 560        This may create groups.
 561
 562        Return:
 563         - Created Groups
 564         - Group Addition Counts
 565        """
 566
 567        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 568        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 569
 570        found_group_memberships, missing_group_memberships, _ = self._resolve_group_memberships(
 571                resolved_course_query.get_id(), resolved_groupset_query.get_id(), memberships, **kwargs)
 572
 573        # Create missing groups.
 574        created_groups = []
 575        for name in sorted(missing_group_memberships.keys()):
 576            group = self.courses_groups_create(resolved_course_query.get_id(), resolved_groupset_query.get_id(), name, **kwargs)
 577            created_groups.append(group)
 578
 579            # Merge in new group with existing structure.
 580            query = group.to_query()
 581            if (query not in found_group_memberships):
 582                found_group_memberships[query] = []
 583
 584            found_group_memberships[query] += missing_group_memberships[name]
 585
 586        # Add memberships.
 587        counts = {}
 588        for resolved_group_query in sorted(found_group_memberships.keys()):
 589            resolved_user_queries = found_group_memberships[resolved_group_query]
 590
 591            count = self.courses_groups_memberships_resolve_and_add(
 592                    resolved_course_query, resolved_groupset_query, resolved_group_query,
 593                    resolved_user_queries,
 594                    **kwargs)
 595
 596            counts[resolved_group_query] = count
 597
 598        return (created_groups, counts)
 599
 600    def courses_groupsets_memberships_resolve_and_set(self,
 601            course_query: lms.model.courses.CourseQuery,
 602            groupset_query: lms.model.groupsets.GroupSetQuery,
 603            memberships: typing.Collection[lms.model.groups.GroupMembership],
 604            **kwargs: typing.Any) -> typing.Tuple[
 605                    typing.List[lms.model.groups.Group],
 606                    typing.List[lms.model.groups.ResolvedGroupQuery],
 607                    typing.Dict[lms.model.groups.ResolvedGroupQuery, int],
 608                    typing.Dict[lms.model.groups.ResolvedGroupQuery, int],
 609            ]:
 610        """
 611        Resolve queries and set the specified group memberships.
 612        This may create and delete groups.
 613
 614        Return:
 615         - Created Groups
 616         - Deleted Groups
 617         - Group Addition Counts
 618         - Group Subtraction Counts
 619        """
 620
 621        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 622        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 623
 624        found_group_memberships, missing_group_memberships, unused_groups = self._resolve_group_memberships(
 625                resolved_course_query.get_id(), resolved_groupset_query.get_id(), memberships, **kwargs)
 626
 627        # Delete unused groups.
 628        deleted_groups = []
 629        for group_query in sorted(unused_groups):
 630            result = self.courses_groups_delete(resolved_course_query.get_id(), resolved_groupset_query.get_id(), group_query.get_id(), **kwargs)
 631            if (result):
 632                deleted_groups.append(group_query)
 633
 634        # Create missing groups.
 635        created_groups = []
 636        for name in sorted(missing_group_memberships.keys()):
 637            group = self.courses_groups_create(resolved_course_query.get_id(), resolved_groupset_query.get_id(), name, **kwargs)
 638            created_groups.append(group)
 639
 640            # Merge in new group with existing structure.
 641            query = group.to_query()
 642            if (query not in found_group_memberships):
 643                found_group_memberships[query] = []
 644
 645            found_group_memberships[query] += missing_group_memberships[name]
 646
 647        # Set memberships.
 648        add_counts = {}
 649        sub_counts = {}
 650        for resolved_group_query in sorted(found_group_memberships.keys()):
 651            resolved_user_queries = found_group_memberships[resolved_group_query]
 652
 653            (add_count, sub_count, deleted) = self.courses_groups_memberships_resolve_and_set(
 654                    resolved_course_query, resolved_groupset_query, resolved_group_query,
 655                    resolved_user_queries,
 656                    delete_empty = True,
 657                    **kwargs)
 658
 659            if (deleted):
 660                deleted_groups.append(resolved_group_query)
 661
 662            add_counts[resolved_group_query] = add_count
 663            sub_counts[resolved_group_query] = sub_count
 664
 665        return (created_groups, deleted_groups, add_counts, sub_counts)
 666
 667    def courses_groupsets_memberships_resolve_and_subtract(self,
 668            course_query: lms.model.courses.CourseQuery,
 669            groupset_query: lms.model.groupsets.GroupSetQuery,
 670            memberships: typing.Collection[lms.model.groups.GroupMembership],
 671            **kwargs: typing.Any) -> typing.Dict[lms.model.groups.ResolvedGroupQuery, int]:
 672        """
 673        Resolve queries and subtract the specified users to the specified groups.
 674        This will not delete any groups.
 675
 676        Return:
 677         - Group Subtraction Counts
 678        """
 679
 680        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 681        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 682
 683        found_group_memberships, missing_group_memberships, _ = self._resolve_group_memberships(
 684                resolved_course_query.get_id(), resolved_groupset_query.get_id(), memberships, **kwargs)
 685
 686        # Warn about missing groups.
 687        for name in sorted(missing_group_memberships.keys()):
 688            _logger.warning("Group does not exist: '%s'.", name)
 689
 690        # Subtract memberships.
 691        counts = {}
 692        for resolved_group_query in sorted(found_group_memberships.keys()):
 693            resolved_user_queries = found_group_memberships[resolved_group_query]
 694
 695            (count, _) = self.courses_groups_memberships_resolve_and_subtract(
 696                    resolved_course_query, resolved_groupset_query, resolved_group_query,
 697                    resolved_user_queries,
 698                    delete_empty = False,
 699                    **kwargs)
 700
 701            counts[resolved_group_query] = count
 702
 703        return counts
 704
 705    def courses_groupsets_memberships_list(self,
 706            course_id: str,
 707            groupset_id: str,
 708            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSetMembership]:
 709        """
 710        List the membership of the group sets associated with the given course.
 711        """
 712
 713        raise NotImplementedError('courses_groupsets_memberships_list')
 714
 715    def courses_groupsets_memberships_resolve_and_list(self,
 716            course_query: lms.model.courses.CourseQuery,
 717            groupset_query: lms.model.groupsets.GroupSetQuery,
 718            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSetMembership]:
 719        """
 720        List the membership of the group sets associated with the given course.
 721        """
 722
 723        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 724        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 725
 726        memberships = self.courses_groupsets_memberships_list(resolved_course_query.get_id(), resolved_groupset_query.get_id(), **kwargs)
 727
 728        # Resolve memberships.
 729
 730        users = self.courses_users_list(resolved_course_query.get_id(), **kwargs)
 731        groups = self.courses_groups_list(resolved_course_query.get_id(), resolved_groupset_query.get_id(), **kwargs)
 732
 733        users_map = {user.id: user.to_query() for user in users}
 734        groups_map = {group.id: group.to_query() for group in groups}
 735
 736        for membership in memberships:
 737            membership.update_queries(resolved_groupset_query, users = users_map, groups = groups_map)
 738
 739        return sorted(memberships)
 740
 741    def courses_groups_create(self,
 742            course_id: str,
 743            groupset_id: str,
 744            name: str,
 745            **kwargs: typing.Any) -> lms.model.groups.Group:
 746        """
 747        Create a group.
 748        """
 749
 750        raise NotImplementedError('courses_groups_create')
 751
 752    def courses_groups_resolve_and_create(self,
 753            course_query: lms.model.courses.CourseQuery,
 754            groupset_query: lms.model.groupsets.GroupSetQuery,
 755            name: str,
 756            **kwargs: typing.Any) -> lms.model.groups.Group:
 757        """
 758        Resolve references and create a group.
 759        """
 760
 761        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 762        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 763        return self.courses_groups_create(resolved_course_query.get_id(), resolved_groupset_query.get_id(), name, **kwargs)
 764
 765    def courses_groups_delete(self,
 766            course_id: str,
 767            groupset_id: str,
 768            group_id: str,
 769            **kwargs: typing.Any) -> bool:
 770        """
 771        Delete a group.
 772        """
 773
 774        raise NotImplementedError('courses_groups_delete')
 775
 776    def courses_groups_resolve_and_delete(self,
 777            course_query: lms.model.courses.CourseQuery,
 778            groupset_query: lms.model.groupsets.GroupSetQuery,
 779            group_query: lms.model.groups.GroupQuery,
 780            **kwargs: typing.Any) -> bool:
 781        """
 782        Resolve references and create a group.
 783        """
 784
 785        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 786        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 787        resolved_group_query = self.resolve_group_query(resolved_course_query.get_id(), resolved_groupset_query.get_id(), group_query, **kwargs)
 788        return self.courses_groups_delete(resolved_course_query.get_id(), resolved_groupset_query.get_id(), resolved_group_query.get_id(), **kwargs)
 789
 790    def courses_groups_get(self,
 791            course_query: lms.model.courses.CourseQuery,
 792            groupset_query: lms.model.groupsets.GroupSetQuery,
 793            group_queries: typing.Collection[lms.model.groups.GroupQuery],
 794            **kwargs: typing.Any) -> typing.List[lms.model.groups.Group]:
 795        """
 796        Get the specified groups associated with the given course.
 797        """
 798
 799        if (len(group_queries) == 0):
 800            return []
 801
 802        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 803        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 804        groups = self.courses_groups_list(resolved_course_query.get_id(), resolved_groupset_query.get_id(), **kwargs)
 805
 806        group_queries = sorted(group_queries)
 807        groups = sorted(groups)
 808
 809        matches = []
 810        for group in groups:
 811            for query in group_queries:
 812                if (query.match(group)):
 813                    matches.append(group)
 814                    break
 815
 816        return matches
 817
 818    def courses_groups_fetch(self,
 819            course_id: str,
 820            groupset_id: str,
 821            group_id: str,
 822            **kwargs: typing.Any) -> typing.Union[lms.model.groups.Group, None]:
 823        """
 824        Fetch a single group associated with the given course.
 825        Return None if no matching group is found.
 826
 827        By default, this will just do a list and choose the relevant record.
 828        Specific backends may override this if there are performance concerns.
 829        """
 830
 831        groups = self.courses_groups_list(course_id, groupset_id, **kwargs)
 832        for group in groups:
 833            if (group.id == group_id):
 834                return group
 835
 836        return None
 837
 838    def courses_groups_list(self,
 839            course_id: str,
 840            groupset_id: str,
 841            **kwargs: typing.Any) -> typing.List[lms.model.groups.Group]:
 842        """
 843        List the groups associated with the given course.
 844        """
 845
 846        raise NotImplementedError('courses_groups_list')
 847
 848    def courses_groups_resolve_and_list(self,
 849            course_query: lms.model.courses.CourseQuery,
 850            groupset_query: lms.model.groupsets.GroupSetQuery,
 851            **kwargs: typing.Any) -> typing.List[lms.model.groups.Group]:
 852        """
 853        List the groups associated with the given course.
 854        """
 855
 856        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 857        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 858        return self.courses_groups_list(resolved_course_query.get_id(), resolved_groupset_query.get_id(), **kwargs)
 859
 860    def courses_groups_memberships_add(self,
 861            course_id: str,
 862            groupset_id: str,
 863            group_id: str,
 864            user_ids: typing.Collection[str],
 865            **kwargs: typing.Any) -> int:
 866        """
 867        Add the specified users to the group.
 868        """
 869
 870        raise NotImplementedError('courses_groups_memberships_add')
 871
 872    def courses_groups_memberships_resolve_and_add(self,
 873            course_query: lms.model.courses.CourseQuery,
 874            groupset_query: lms.model.groupsets.GroupSetQuery,
 875            group_query: lms.model.groups.GroupQuery,
 876            user_queries: typing.Collection[lms.model.users.UserQuery],
 877            **kwargs: typing.Any) -> int:
 878        """
 879        Resolve queries and add the specified users to the group.
 880        """
 881
 882        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 883        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 884        resolved_group_query = self.resolve_group_query(resolved_course_query.get_id(), resolved_groupset_query.get_id(), group_query, **kwargs)
 885        resolved_user_queries = self.resolve_user_queries(resolved_course_query.get_id(), user_queries, warn_on_miss = True, **kwargs)
 886
 887        # Get users already in this group.
 888        group_memberships = self.courses_groups_memberships_list(
 889                resolved_course_query.get_id(),
 890                resolved_groupset_query.get_id(),
 891                resolved_group_query.get_id(),
 892                **kwargs)
 893
 894        group_user_ids = {membership.user.id for membership in group_memberships if membership.user.id is not None}
 895
 896        # Filter out users already in the group.
 897        user_ids = []
 898        for query in sorted(resolved_user_queries):
 899            if (query.get_id() in group_user_ids):
 900                _logger.warning("User '%s' already in group '%s'.", query, resolved_group_query)
 901                continue
 902
 903            user_ids.append(query.get_id())
 904
 905        if (len(user_ids) == 0):
 906            return 0
 907
 908        return self.courses_groups_memberships_add(
 909                resolved_course_query.get_id(),
 910                resolved_groupset_query.get_id(),
 911                resolved_group_query.get_id(),
 912                user_ids,
 913                **kwargs)
 914
 915    def courses_groups_memberships_list(self,
 916            course_id: str,
 917            groupset_id: str,
 918            group_id: str,
 919            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSetMembership]:
 920        """
 921        List the membership of the group associated with the given group set.
 922        """
 923
 924        raise NotImplementedError('courses_groups_memberships_list')
 925
 926    def courses_groups_memberships_resolve_and_list(self,
 927            course_query: lms.model.courses.CourseQuery,
 928            groupset_query: lms.model.groupsets.GroupSetQuery,
 929            group_query: lms.model.groups.GroupQuery,
 930            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.GroupSetMembership]:
 931        """
 932        List the membership of the group associated with the given group set.
 933        """
 934
 935        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 936        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 937
 938        groups = self.courses_groups_get(resolved_course_query, resolved_groupset_query, [group_query], **kwargs)
 939        if (len(groups) == 0):
 940            raise ValueError(f"Unable to find group: '{group_query}'.")
 941
 942        group = groups[0]
 943
 944        memberships = self.courses_groups_memberships_list(
 945                resolved_course_query.get_id(),
 946                resolved_groupset_query.get_id(),
 947                group.id,
 948                **kwargs)
 949
 950        # Resolve memberships.
 951
 952        users = self.courses_users_list(resolved_course_query.get_id(), **kwargs)
 953        users_map = {user.id: user.to_query() for user in users}
 954
 955        groups_map = {group.id: group.to_query()}
 956
 957        for membership in memberships:
 958            membership.update_queries(resolved_groupset_query, users = users_map, groups = groups_map)
 959
 960        return sorted(memberships)
 961
 962    def courses_groups_memberships_resolve_and_set(self,
 963            course_query: lms.model.courses.CourseQuery,
 964            groupset_query: lms.model.groupsets.GroupSetQuery,
 965            group_query: lms.model.groups.GroupQuery,
 966            user_queries: typing.Collection[lms.model.users.UserQuery],
 967            delete_empty: bool = False,
 968            **kwargs: typing.Any) -> typing.Tuple[int, int, bool]:
 969        """
 970        Resolve queries and set the specified users for the group.
 971        This method can both add and subtract users from the group.
 972
 973        Returns:
 974         - The count of users added.
 975         - The count of users subtracted.
 976         - If this group was deleted.
 977        """
 978
 979        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
 980        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
 981        resolved_group_query = self.resolve_group_query(resolved_course_query.get_id(), resolved_groupset_query.get_id(), group_query, **kwargs)
 982        resolved_user_queries = self.resolve_user_queries(resolved_course_query.get_id(), user_queries, warn_on_miss = True, **kwargs)
 983
 984        # Get users already in this group.
 985        group_memberships = self.courses_groups_memberships_list(
 986                resolved_course_query.get_id(),
 987                resolved_groupset_query.get_id(),
 988                resolved_group_query.get_id(),
 989                **kwargs)
 990
 991        group_user_queries = {membership.user for membership in group_memberships if membership.user is not None}
 992        group_user_ids = {membership.user.id for membership in group_memberships if membership.user.id is not None}
 993        query_user_ids = {resolved_user_query.get_id() for resolved_user_query in resolved_user_queries}
 994
 995        # Collect users that need to be added.
 996        add_user_ids = []
 997        for query_user_id in query_user_ids:
 998            if (query_user_id not in group_user_ids):
 999                add_user_ids.append(query_user_id)
1000
1001        # Collect users that need to be subtracted.
1002        sub_user_queries = []
1003        for group_user_query in group_user_queries:
1004            if (group_user_query not in resolved_user_queries):
1005                sub_user_queries.append(group_user_query)
1006
1007        # Update the group.
1008
1009        add_count = 0
1010        if (len(add_user_ids) > 0):
1011            add_count = self.courses_groups_memberships_add(
1012                    resolved_course_query.get_id(),
1013                    resolved_groupset_query.get_id(),
1014                    resolved_group_query.get_id(),
1015                    add_user_ids,
1016                    **kwargs)
1017
1018        sub_count = 0
1019        deleted = False
1020        if (len(sub_user_queries) > 0):
1021            sub_count, deleted = self.courses_groups_memberships_resolve_and_subtract(
1022                    resolved_course_query,
1023                    resolved_groupset_query,
1024                    resolved_group_query,
1025                    sub_user_queries,
1026                    delete_empty = delete_empty,
1027                    **kwargs)
1028
1029        return add_count, sub_count, deleted
1030
1031    def courses_groups_memberships_subtract(self,
1032            course_id: str,
1033            groupset_id: str,
1034            group_id: str,
1035            user_ids: typing.Collection[str],
1036            **kwargs: typing.Any) -> int:
1037        """
1038        Subtract the specified users from the group.
1039        """
1040
1041        raise NotImplementedError('courses_groups_memberships_subtract')
1042
1043    def courses_groups_memberships_resolve_and_subtract(self,
1044            course_query: lms.model.courses.CourseQuery,
1045            groupset_query: lms.model.groupsets.GroupSetQuery,
1046            group_query: lms.model.groups.GroupQuery,
1047            user_queries: typing.Collection[lms.model.users.UserQuery],
1048            delete_empty: bool = False,
1049            **kwargs: typing.Any) -> typing.Tuple[int, bool]:
1050        """
1051        Resolve queries and subtract the specified users from the group.
1052        Return:
1053            - The number of users deleted.
1054            - If this group was deleted.
1055        """
1056
1057        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
1058        resolved_groupset_query = self.resolve_groupset_query(resolved_course_query.get_id(), groupset_query, **kwargs)
1059        resolved_group_query = self.resolve_group_query(resolved_course_query.get_id(), resolved_groupset_query.get_id(), group_query, **kwargs)
1060        resolved_user_queries = self.resolve_user_queries(resolved_course_query.get_id(), user_queries, warn_on_miss = True, **kwargs)
1061
1062        # Get users already in this group.
1063        group_memberships = self.courses_groups_memberships_list(
1064                resolved_course_query.get_id(),
1065                resolved_groupset_query.get_id(),
1066                resolved_group_query.get_id(),
1067                **kwargs)
1068
1069        group_user_ids = {membership.user.id for membership in group_memberships if membership.user.id is not None}
1070
1071        # Filter out users not in the group.
1072        user_ids = []
1073        for query in resolved_user_queries:
1074            if (query.get_id() not in group_user_ids):
1075                _logger.warning("User '%s' is not in group '%s'.", query, resolved_group_query)
1076                continue
1077
1078            user_ids.append(query.get_id())
1079
1080        if (delete_empty and len(group_memberships) == 0):
1081            deleted = self.courses_groups_delete(
1082                resolved_course_query.get_id(),
1083                resolved_groupset_query.get_id(),
1084                resolved_group_query.get_id(),
1085                **kwargs)
1086            return 0, deleted
1087
1088        if (len(user_ids) == 0):
1089            return 0, False
1090
1091        count = self.courses_groups_memberships_subtract(
1092                resolved_course_query.get_id(),
1093                resolved_groupset_query.get_id(),
1094                resolved_group_query.get_id(),
1095                user_ids,
1096                **kwargs)
1097
1098        deleted = False
1099        if (delete_empty and (count == len(group_memberships))):
1100            deleted = self.courses_groups_delete(
1101                resolved_course_query.get_id(),
1102                resolved_groupset_query.get_id(),
1103                resolved_group_query.get_id(),
1104                **kwargs)
1105
1106        return count, deleted
1107
1108    def courses_syllabus_fetch(self,
1109            course_id: str,
1110            **kwargs: typing.Any) -> typing.Union[str, None]:
1111        """
1112        Get the syllabus for a course, or None if no syllabus exists.
1113        """
1114
1115        raise NotImplementedError('courses_syllabus_fetch')
1116
1117    def courses_syllabus_get(self,
1118            course_query: lms.model.courses.CourseQuery,
1119            **kwargs: typing.Any) -> typing.Union[str, None]:
1120        """
1121        Get the syllabus for a course query, or None if no syllabus exists.
1122        """
1123
1124        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
1125
1126        return self.courses_syllabus_fetch(resolved_course_query.get_id(), **kwargs)
1127
1128    def courses_users_get(self,
1129            course_query: lms.model.courses.CourseQuery,
1130            user_queries: typing.Collection[lms.model.users.UserQuery],
1131            **kwargs: typing.Any) -> typing.List[lms.model.users.CourseUser]:
1132        """
1133        Get the specified users associated with the given course.
1134        """
1135
1136        if (len(user_queries) == 0):
1137            return []
1138
1139        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
1140        users = self.courses_users_list(resolved_course_query.get_id(), **kwargs)
1141
1142        user_queries = sorted(user_queries)
1143        users = sorted(users)
1144
1145        matches = []
1146        for user in users:
1147            for query in user_queries:
1148                if (query.match(user)):
1149                    matches.append(user)
1150                    break
1151
1152        return matches
1153
1154    def courses_users_fetch(self,
1155            course_id: str,
1156            user_id: str,
1157            **kwargs: typing.Any) -> typing.Union[lms.model.users.CourseUser, None]:
1158        """
1159        Fetch a single user associated with the given course.
1160        Return None if no matching user is found.
1161
1162        By default, this will just do a list and choose the relevant record.
1163        Specific backends may override this if there are performance concerns.
1164        """
1165
1166        users = self.courses_users_list(course_id, **kwargs)
1167        for user in sorted(users):
1168            if (user.id == user_id):
1169                return user
1170
1171        return None
1172
1173    def courses_users_list(self,
1174            course_id: str,
1175            **kwargs: typing.Any) -> typing.List[lms.model.users.CourseUser]:
1176        """
1177        List the users associated with the given course.
1178        """
1179
1180        raise NotImplementedError('courses_users_list')
1181
1182    def courses_users_resolve_and_list(self,
1183            course_query: lms.model.courses.CourseQuery,
1184            **kwargs: typing.Any) -> typing.List[lms.model.users.CourseUser]:
1185        """
1186        List the users associated with the given course.
1187        """
1188
1189        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
1190        return list(sorted(self.courses_users_list(resolved_course_query.get_id(), **kwargs)))
1191
1192    def courses_users_scores_get(self,
1193            course_query: lms.model.courses.CourseQuery,
1194            user_query: lms.model.users.UserQuery,
1195            assignment_queries: typing.Collection[lms.model.assignments.AssignmentQuery],
1196            **kwargs: typing.Any) -> typing.List[lms.model.scores.AssignmentScore]:
1197        """
1198        Get the scores associated with the given user query and assignment queries.
1199        """
1200
1201        if (len(assignment_queries) == 0):
1202            return []
1203
1204        scores = self.courses_users_scores_resolve_and_list(course_query, user_query, **kwargs)
1205
1206        scores = sorted(scores)
1207        assignment_queries = sorted(assignment_queries)
1208
1209        matches = []
1210        for score in scores:
1211            for assignment_query in assignment_queries:
1212                if (assignment_query.match(score.assignment)):
1213                    matches.append(score)
1214
1215        return matches
1216
1217    def courses_users_scores_fetch(self,
1218            course_id: str,
1219            user_id: str,
1220            assignment_id: str,
1221            **kwargs: typing.Any) -> typing.Union[lms.model.scores.AssignmentScore, None]:
1222        """
1223        Fetch the score associated with the given user and assignment.
1224
1225        By default, this will just do a list and choose the relevant record.
1226        Specific backends may override this if there are performance concerns.
1227        """
1228
1229        # The default implementation is the same as courses_assignments_scores_fetch().
1230        return self.courses_assignments_scores_fetch(course_id, assignment_id, user_id, **kwargs)
1231
1232    def courses_users_scores_list(self,
1233            course_id: str,
1234            user_id: str,
1235            **kwargs: typing.Any) -> typing.List[lms.model.scores.AssignmentScore]:
1236        """
1237        List the scores associated with the given user.
1238        """
1239
1240        raise NotImplementedError('courses_users_scores_list')
1241
1242    def courses_users_scores_resolve_and_list(self,
1243            course_query: lms.model.courses.CourseQuery,
1244            user_query: lms.model.users.UserQuery,
1245            **kwargs: typing.Any) -> typing.List[lms.model.scores.AssignmentScore]:
1246        """
1247        List the scores associated with the given user query.
1248        In addition to resolving the user query,
1249        assignments will also be resolved into their full version
1250        (instead of the reduced version usually returned with scores).
1251        """
1252
1253        resolved_course_query = self.resolve_course_query(course_query, **kwargs)
1254
1255        # Resolve the user query.
1256        matched_users = self.courses_users_get(resolved_course_query, [user_query], **kwargs)
1257        if (len(matched_users) == 0):
1258            return []
1259
1260        target_user = matched_users[0]
1261
1262        # List the scores.
1263        scores = self.courses_users_scores_list(resolved_course_query.get_id(), target_user.id, **kwargs)
1264        if (len(scores) == 0):
1265            return []
1266
1267        # Resolve the scores' queries.
1268
1269        assignments = self.courses_assignments_list(resolved_course_query.get_id(), **kwargs)
1270        assignments_map = {assignment.id: assignment for assignment in assignments}
1271
1272        for score in scores:
1273            score.user = target_user.to_query()
1274
1275            if ((score.assignment is not None) and (score.assignment.id in assignments_map)):
1276                score.assignment = assignments_map[score.assignment.id].to_query()
1277
1278        return sorted(scores)
1279
1280    # Utility Methods
1281
1282    def parse_assignment_query(self, text: typing.Union[str, None]) -> typing.Union[lms.model.assignments.AssignmentQuery, None]:
1283        """
1284        Attempt to parse an assignment query from a string.
1285        The there is no query, return a None.
1286        If the query is malformed, raise an exception.
1287
1288        By default, this method assumes that LMS IDs are ints.
1289        Child backends may override this to implement their specific behavior.
1290        """
1291
1292        return lms.model.query.parse_int_query(lms.model.assignments.AssignmentQuery, text, check_email = False)
1293
1294    def parse_assignment_queries(self, texts: typing.Collection[typing.Union[str, None]]) -> typing.List[lms.model.assignments.AssignmentQuery]:
1295        """ Parse a list of assignment queries. """
1296
1297        queries = []
1298        for text in texts:
1299            query = self.parse_assignment_query(text)
1300            if (query is not None):
1301                queries.append(query)
1302
1303        return queries
1304
1305    def parse_course_query(self, text: typing.Union[str, None]) -> typing.Union[lms.model.courses.CourseQuery, None]:
1306        """
1307        Attempt to parse a course query from a string.
1308        The there is no query, return a None.
1309        If the query is malformed, raise an exception.
1310
1311        By default, this method assumes that LMS IDs are ints.
1312        Child backends may override this to implement their specific behavior.
1313        """
1314
1315        return lms.model.query.parse_int_query(lms.model.courses.CourseQuery, text, check_email = False)
1316
1317    def parse_course_queries(self, texts: typing.Collection[typing.Union[str, None]]) -> typing.List[lms.model.courses.CourseQuery]:
1318        """ Parse a list of course queries. """
1319
1320        queries = []
1321        for text in texts:
1322            query = self.parse_course_query(text)
1323            if (query is not None):
1324                queries.append(query)
1325
1326        return queries
1327
1328    def parse_groupset_query(self, text: typing.Union[str, None]) -> typing.Union[lms.model.groupsets.GroupSetQuery, None]:
1329        """
1330        Attempt to parse a group set query from a string.
1331        The there is no query, return a None.
1332        If the query is malformed, raise an exception.
1333
1334        By default, this method assumes that LMS IDs are ints.
1335        Child backends may override this to implement their specific behavior.
1336        """
1337
1338        return lms.model.query.parse_int_query(lms.model.groupsets.GroupSetQuery, text, check_email = False)
1339
1340    def parse_groupset_queries(self, texts: typing.Collection[typing.Union[str, None]]) -> typing.List[lms.model.groupsets.GroupSetQuery]:
1341        """ Parse a list of group set queries. """
1342
1343        queries = []
1344        for text in texts:
1345            query = self.parse_groupset_query(text)
1346            if (query is not None):
1347                queries.append(query)
1348
1349        return queries
1350
1351    def parse_group_query(self, text: typing.Union[str, None]) -> typing.Union[lms.model.groups.GroupQuery, None]:
1352        """
1353        Attempt to parse a group query from a string.
1354        The there is no query, return a None.
1355        If the query is malformed, raise an exception.
1356
1357        By default, this method assumes that LMS IDs are ints.
1358        Child backends may override this to implement their specific behavior.
1359        """
1360
1361        return lms.model.query.parse_int_query(lms.model.groups.GroupQuery, text, check_email = False)
1362
1363    def parse_group_queries(self, texts: typing.Collection[typing.Union[str, None]]) -> typing.List[lms.model.groups.GroupQuery]:
1364        """ Parse a list of group queries. """
1365
1366        queries = []
1367        for text in texts:
1368            query = self.parse_group_query(text)
1369            if (query is not None):
1370                queries.append(query)
1371
1372        return queries
1373
1374    def parse_user_query(self, text: typing.Union[str, None]) -> typing.Union[lms.model.users.UserQuery, None]:
1375        """
1376        Attempt to parse a user query from a string.
1377        The there is no query, return a None.
1378        If the query is malformed, raise an exception.
1379
1380        By default, this method assumes that LMS IDs are ints.
1381        Child backends may override this to implement their specific behavior.
1382        """
1383
1384        return lms.model.query.parse_int_query(lms.model.users.UserQuery, text, check_email = True)
1385
1386    def parse_user_queries(self, texts: typing.Collection[typing.Union[str, None]]) -> typing.List[lms.model.users.UserQuery]:
1387        """ Parse a list of user queries. """
1388
1389        queries = []
1390        for text in texts:
1391            query = self.parse_user_query(text)
1392            if (query is not None):
1393                queries.append(query)
1394
1395        return queries
1396
1397    def resolve_assignment_query(self,
1398            course_id: str,
1399            assignment_query: lms.model.assignments.AssignmentQuery,
1400            **kwargs: typing.Any) -> lms.model.assignments.ResolvedAssignmentQuery:
1401        """ Resolve the assignment query or raise an exception. """
1402
1403        # Shortcut already resolved queries.
1404        if (isinstance(assignment_query, lms.model.assignments.ResolvedAssignmentQuery)):
1405            return assignment_query
1406
1407        results = self.resolve_assignment_queries(course_id, [assignment_query], **kwargs)
1408        if (len(results) == 0):
1409            raise ValueError(f"Could not resolve assignment query: '{assignment_query}'.")
1410
1411        return results[0]
1412
1413    def resolve_assignment_queries(self,
1414            course_id: str,
1415            queries: typing.Collection[lms.model.assignments.AssignmentQuery],
1416            **kwargs: typing.Any) -> typing.List[lms.model.assignments.ResolvedAssignmentQuery]:
1417        """
1418        Resolve a list of assignment queries into a list of resolved assignment queries.
1419        See _resolve_queries().
1420        """
1421
1422        results = self._resolve_queries(
1423            queries,
1424            'assignment',
1425            self.courses_assignments_list(course_id, **kwargs),
1426            lms.model.assignments.ResolvedAssignmentQuery,
1427            **kwargs)
1428
1429        return typing.cast(typing.List[lms.model.assignments.ResolvedAssignmentQuery], results)
1430
1431    def resolve_course_query(self,
1432            query: lms.model.courses.CourseQuery,
1433            **kwargs: typing.Any) -> lms.model.courses.ResolvedCourseQuery:
1434        """ Resolve the course query or raise an exception. """
1435
1436        # Shortcut already resolved queries.
1437        if (isinstance(query, lms.model.courses.ResolvedCourseQuery)):
1438            return query
1439
1440        results = self.resolve_course_queries([query], **kwargs)
1441        if (len(results) == 0):
1442            raise ValueError(f"Could not resolve course query: '{query}'.")
1443
1444        return results[0]
1445
1446    def resolve_course_queries(self,
1447            queries: typing.Collection[lms.model.courses.CourseQuery],
1448            **kwargs: typing.Any) -> typing.List[lms.model.courses.ResolvedCourseQuery]:
1449        """
1450        Resolve a list of course queries into a list of resolved course queries.
1451        See _resolve_queries().
1452        """
1453
1454        results = self._resolve_queries(
1455            queries,
1456            'course',
1457            self.courses_list(**kwargs),
1458            lms.model.courses.ResolvedCourseQuery,
1459            **kwargs)
1460
1461        return typing.cast(typing.List[lms.model.courses.ResolvedCourseQuery], results)
1462
1463    def resolve_group_queries(self,
1464            course_id: str,
1465            groupset_id: str,
1466            queries: typing.Collection[lms.model.groups.GroupQuery],
1467            **kwargs: typing.Any) -> typing.List[lms.model.groups.ResolvedGroupQuery]:
1468        """
1469        Resolve a list of group queries into a list of resolved group queries.
1470        See _resolve_queries().
1471        """
1472
1473        results = self._resolve_queries(
1474            queries,
1475            'group',
1476            self.courses_groups_list(course_id, groupset_id, **kwargs),
1477            lms.model.groups.ResolvedGroupQuery,
1478            **kwargs)
1479
1480        return typing.cast(typing.List[lms.model.groups.ResolvedGroupQuery], results)
1481
1482    def resolve_group_query(self,
1483            course_id: str,
1484            groupset_id: str,
1485            query: lms.model.groups.GroupQuery,
1486            **kwargs: typing.Any) -> lms.model.groups.ResolvedGroupQuery:
1487        """ Resolve the group query or raise an exception. """
1488
1489        # Shortcut already resolved queries.
1490        if (isinstance(query, lms.model.groups.ResolvedGroupQuery)):
1491            return query
1492
1493        results = self.resolve_group_queries(course_id, groupset_id, [query], **kwargs)
1494        if (len(results) == 0):
1495            raise ValueError(f"Could not resolve group query: '{query}'.")
1496
1497        return results[0]
1498
1499    def resolve_groupset_queries(self,
1500            course_id: str,
1501            queries: typing.Collection[lms.model.groupsets.GroupSetQuery],
1502            **kwargs: typing.Any) -> typing.List[lms.model.groupsets.ResolvedGroupSetQuery]:
1503        """
1504        Resolve a list of group set queries into a list of resolved group set queries.
1505        See _resolve_queries().
1506        """
1507
1508        results = self._resolve_queries(
1509            queries,
1510            'group set',
1511            self.courses_groupsets_list(course_id, **kwargs),
1512            lms.model.groupsets.ResolvedGroupSetQuery,
1513            **kwargs)
1514
1515        return typing.cast(typing.List[lms.model.groupsets.ResolvedGroupSetQuery], results)
1516
1517    def resolve_groupset_query(self,
1518            course_id: str,
1519            groupset_query: lms.model.groupsets.GroupSetQuery,
1520            **kwargs: typing.Any) -> lms.model.groupsets.ResolvedGroupSetQuery:
1521        """ Resolve the group set query or raise an exception. """
1522
1523        # Shortcut already resolved queries.
1524        if (isinstance(groupset_query, lms.model.groupsets.ResolvedGroupSetQuery)):
1525            return groupset_query
1526
1527        results = self.resolve_groupset_queries(course_id, [groupset_query], **kwargs)
1528        if (len(results) == 0):
1529            raise ValueError(f"Could not resolve group set query: '{groupset_query}'.")
1530
1531        return results[0]
1532
1533    def resolve_user_queries(self,
1534            course_id: str,
1535            queries: typing.Collection[lms.model.users.UserQuery],
1536            only_students: bool = False,
1537            **kwargs: typing.Any) -> typing.List[lms.model.users.ResolvedUserQuery]:
1538        """
1539        Resolve a list of user queries into a list of resolved user queries.
1540        See _resolve_queries().
1541        """
1542
1543        filter_func = None
1544        if (only_students):
1545            filter_func = lambda user: user.is_student()  # pylint: disable=unnecessary-lambda-assignment
1546
1547        results = self._resolve_queries(
1548            queries,
1549            'user',
1550            self.courses_users_list(course_id, **kwargs),
1551            lms.model.users.ResolvedUserQuery,
1552            filter_func = filter_func,
1553            **kwargs)
1554
1555        return typing.cast(typing.List[lms.model.users.ResolvedUserQuery], results)
1556
1557    def _resolve_queries(self,
1558            queries: typing.Collection[lms.model.query.BaseQuery],
1559            label: str,
1560            items: typing.Collection,
1561            resolved_query_class: typing.Type,
1562            empty_all: bool = False,
1563            warn_on_miss: bool = False,
1564            filter_func: typing.Union[typing.Callable, None] = None,
1565            **kwargs: typing.Any) -> typing.List[lms.model.query.ResolvedBaseQuery]:
1566        """
1567        Resolve a list of queries.
1568        The returned list may be shorter than the list of queries (if input queries are not matched).
1569        The queries will be deduplicated and sorted.
1570
1571        If |empty_all| is true and no queries are specified, then all items will be returned.
1572
1573        If |filter_func| is passed, then that function will be called with each raw item,
1574        and ones that return true will be kept.
1575        """
1576
1577        if (filter_func is not None):
1578            items = list(filter(filter_func, items))
1579
1580        if (empty_all and (len(queries) == 0)):
1581            return list(sorted({resolved_query_class(item) for item in items}))
1582
1583        matched_queries = []  # type: ignore[var-annotated]
1584        for query in queries:
1585            match = False
1586            for item in items:
1587                if (query.match(item)):
1588                    matched_query = resolved_query_class(item)
1589
1590                    if (match):
1591                        raise ValueError(
1592                            f"Ambiguous {label} query ('{query}')"
1593                            f" matches multiple {label}s ['{matched_queries[-1]}', '{matched_query}'].")
1594
1595                    matched_queries.append(matched_query)
1596                    match = True
1597
1598            if ((not match) and warn_on_miss):
1599                _logger.warning("Could not resolve %s query '%s'.", label, query)
1600
1601        return list(sorted(set(matched_queries)))
1602
1603    def _resolve_group_memberships(self,
1604            course_id: str,
1605            groupset_id: str,
1606            memberships: typing.Collection[lms.model.groups.GroupMembership],
1607            **kwargs: typing.Any) -> typing.Tuple[
1608                typing.Dict[lms.model.groups.ResolvedGroupQuery, typing.List[lms.model.users.ResolvedUserQuery]],
1609                typing.Dict[str, typing.List[lms.model.users.ResolvedUserQuery]],
1610                typing.List[lms.model.groups.ResolvedGroupQuery]]:
1611        """
1612        Resolve a list of group memberships.
1613        This method will resolved each query and split up the memberships by the appropriate group.
1614        If a group does not exist, the memberships will be split by apparent group name.
1615
1616        Returns:
1617         - Memberships in Found Groups (keyed by resolved group query)
1618         - Memberships in Missing Groups (keyed by apparent group name)
1619         - Groups not involved in any of the returned memberships.
1620
1621        The returned dicts will be the found groups (keyed by resolved query) and then the missing groups (keyed by apparent group name).
1622        """
1623
1624        found_group_memberships: typing.Dict[lms.model.groups.ResolvedGroupQuery, typing.List[lms.model.users.ResolvedUserQuery]] = {}
1625        missing_group_memberships: typing.Dict[str, typing.List[lms.model.users.ResolvedUserQuery]] = {}
1626
1627        users = self.courses_users_list(course_id, **kwargs)
1628        resolved_user_queries = [user.to_query() for user in sorted(users)]
1629
1630        groups = self.courses_groups_list(course_id, groupset_id, **kwargs)
1631        resolved_group_queries = [group.to_query() for group in sorted(groups)]
1632
1633        for (i, membership) in enumerate(memberships):
1634            # Resolve user.
1635
1636            resolved_user_query = None
1637            for possible_user_query in resolved_user_queries:
1638                if (membership.user.match(possible_user_query)):
1639                    resolved_user_query = possible_user_query
1640                    break
1641
1642            if (resolved_user_query is None):
1643                _logger.warning("Could not resolve user '%s' for membership entry at index %d.", membership.user, i)
1644                continue
1645
1646            # Resolve group.
1647
1648            resolved_group_query = None
1649            for possible_group_query in resolved_group_queries:
1650                if (membership.group.match(possible_group_query)):
1651                    resolved_group_query = possible_group_query
1652                    break
1653
1654            # Add to the correct collection.
1655
1656            if (resolved_group_query is None):
1657                if ((membership.group.name is None) or (len(membership.group.name) == 0)):
1658                    _logger.warning(("Membership entry at index %d has a group with no name."
1659                        + " Ensure that non-existent groups all have names."), i)
1660                    continue
1661
1662                if (membership.group.name not in missing_group_memberships):
1663                    missing_group_memberships[membership.group.name] = []
1664
1665                missing_group_memberships[membership.group.name].append(resolved_user_query)
1666            else:
1667                if (resolved_group_query not in found_group_memberships):
1668                    found_group_memberships[resolved_group_query] = []
1669
1670                found_group_memberships[resolved_group_query].append(resolved_user_query)
1671
1672        unused_groups = sorted(list(set(resolved_group_queries) - set(found_group_memberships.keys())))
1673
1674        return (found_group_memberships, missing_group_memberships, unused_groups)