lms.model.backend

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