Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __create_examiner_objects(self, groupqueryset):
examiners = []
groupcount = 0
candidatecount = 0
for group in groupqueryset:
examiner = Examiner(assignmentgroup=group,
relatedexaminer=self.get_relatedexaminer())
examiners.append(examiner)
groupcount += 1
candidatecount += len(group.candidates.all())
Examiner.objects.bulk_create(examiners)
return groupcount, candidatecount
def start_migration(self, from_db_object):
relatedexaminer = self.get_relatedexaminer(
user_shortname=from_db_object.relatedexaminer.user.shortname,
period_shortname=from_db_object.relatedexaminer.period.short_name,
subject_shortname=from_db_object.relatedexaminer.period.parentnode.short_name)
assignment_group = self.__get_assignment_group_from_temp_merge_id(from_db_obj=from_db_object)
if relatedexaminer and assignment_group:
examiner_kwargs = model_to_dict(from_db_object, exclude=[
'id', 'pk', 'assignmentgroup', 'relatedexaminer', 'old_reference_not_in_use_user'])
examiner = core_models.Examiner(**examiner_kwargs)
examiner.assignmentgroup_id = assignment_group.id
examiner.relatedexaminer_id = relatedexaminer.id
self.save_object(obj=examiner)
else:
raise ValueError('RelatedExaminers and AssignmentGroups must be imported before Candidates')
def __random_organize_examiners(self, groupqueryset, relatedexaminerqueryset):
relatedexaminer_ids = list(relatedexaminerqueryset.values_list('id', flat=True))
group_ids = list(groupqueryset.values_list('id', flat=True))
random.shuffle(relatedexaminer_ids)
random.shuffle(group_ids)
examiners_to_create = []
while group_ids:
for relatedexaminer in relatedexaminer_ids:
if not group_ids:
break
examiners_to_create.append(
Examiner(relatedexaminer_id=relatedexaminer, assignmentgroup_id=group_ids.pop(0))
)
Examiner.objects.bulk_create(examiners_to_create)
relatedstudent_to_group_map[candidate.relatedstudent_id] = [candidate.assignment_group]
else:
relatedstudent_to_group_map[candidate.relatedstudent_id].append(candidate.assignment_group)
examiners_to_create = []
for periodtag in period_tag_queryset:
group_list = []
for relatedstudent in periodtag.relatedstudents.all():
relatedstudent_id = relatedstudent.id
if relatedstudent_id in relatedstudent_to_group_map:
group_list.extend(relatedstudent_to_group_map[relatedstudent_id])
del relatedstudent_to_group_map[relatedstudent_id]
for relatedexaminer in periodtag.relatedexaminers.all():
for group in group_list:
examiners_to_create.append(Examiner(assignmentgroup=group, relatedexaminer=relatedexaminer))
Examiner.objects.bulk_create(examiners_to_create)
def __random_organize_examiners(self, groupqueryset, relatedexaminerqueryset):
relatedexaminer_ids = list(relatedexaminerqueryset.values_list('id', flat=True))
group_ids = list(groupqueryset.values_list('id', flat=True))
random.shuffle(relatedexaminer_ids)
random.shuffle(group_ids)
examiners_to_create = []
while group_ids:
for relatedexaminer in relatedexaminer_ids:
if not group_ids:
break
examiners_to_create.append(
Examiner(relatedexaminer_id=relatedexaminer, assignmentgroup_id=group_ids.pop(0))
)
Examiner.objects.bulk_create(examiners_to_create)
def prepare_examiner_ids(self, obj):
return [
examiner.user.id
for examiner in models.Examiner.objects.filter(assignmentgroup__parentnode=obj.id)]
def __create_examiner_objects(self, groupqueryset):
examiners = []
groupcount = 0
candidatecount = 0
for group in groupqueryset:
examiner = Examiner(assignmentgroup=group,
relatedexaminer=self.get_relatedexaminer())
examiners.append(examiner)
groupcount += 1
candidatecount += len(group.candidates.all())
Examiner.objects.bulk_create(examiners)
return groupcount, candidatecount
def __get_default_db_model_count_map(self, migrate_from_alias):
return {
'assignment_group': AssignmentGroup.objects.count(),
'candidate': Candidate.objects.count(),
'examiner': Examiner.objects.count(),
'feedback_set': FeedbackSet.objects.count(),
'comment': Comment.objects.count(),
'comment_file': CommentFile.objects.count(),
'group_comment': GroupComment.objects.count()
}
def _get_examinerqueryset(self):
"""Get examiners.
Returns:
QuerySet: A queryset of :class:`~devilry.apps.core.models.Examiner`s.
"""
return Examiner.objects \
.select_related('relatedexaminer') \
.order_by(
Lower(Concat('relatedexaminer__user__fullname',
'relatedexaminer__user__shortname')))
# Step2: Bulk create candidate and examiners from group.copied_from..
candidates = []
examiners = []
for group in self.assignmentgroups \
.prefetch_related(
models.Prefetch(
'copied_from',
to_attr='copied_from_list',
queryset=AssignmentGroup.objects.prefetch_related(
models.Prefetch('candidates',
to_attr='candidatelist',
queryset=Candidate.objects.all()),
models.Prefetch('examiners',
to_attr='examinerlist',
queryset=Examiner.objects.all()),
)
)
):
for othercandidate in group.copied_from_list.candidatelist:
newcandidate = Candidate(
assignment_group=group,
relatedstudent_id=othercandidate.relatedstudent_id,
)
candidates.append(newcandidate)
for otherexaminer in group.copied_from_list.examinerlist:
newexaminer = Examiner(
assignmentgroup=group,
relatedexaminer_id=otherexaminer.relatedexaminer_id
)
examiners.append(newexaminer)
Candidate.objects.bulk_create(candidates)