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