Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Candidates
if len(pathsplit) > 3:
usernames = pathsplit[3].split(',')
users = []
for u in usernames:
try:
user = get_user_model().objects.get(shortname=u)
except get_user_model().DoesNotExist:
user = get_user_model().objects.create_user(username=u)
users.append(user)
assignment_group = AssignmentGroup(parentnode=assignment)
assignment_group.clean()
assignment_group.save()
for user in users:
assignment_group.candidates.add(Candidate(student=user))
last = assignment_group
return last
def add_students(self, *users):
candidates = [Candidate(student=user) for user in users]
self.add_candidates(*candidates)
return self
def handle(self, *args, **kwargs):
from devilry.apps.core.models import Candidate
verbosity = int(kwargs.get('verbosity', '1'))
updates = 0
with transaction.commit_manually():
for candidate in Candidate.objects.all():
candidate.save()
if verbosity > 1:
print 'Updated {0}'.format(candidate)
updates += 1
transaction.commit()
if verbosity > 0:
print 'Successfully updated {0} candidates.'.format(updates)
for candidate in to_delete:
candidate.delete()
create_kwargs = []
update_candidates = []
for candidatespec in obj.fake_candidates:
username = candidatespec['username']
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
raise InvalidUsername(username)
else:
candidate_id = candidatespec.get('candidate_id', None)
try:
candiate = obj.candidates.get(student__username=username)
except models.Candidate.DoesNotExist:
candidatekwargs = dict(student = user,
candidate_id = candidate_id)
create_kwargs.append(candidatekwargs)
else:
update_candidates.append((candiate, candidate_id))
for candidate, candidate_id in update_candidates:
candidate.candidate_id = candidate_id
candidate.save()
for candidatekwargs in create_kwargs:
obj.candidates.create(**candidatekwargs)
def recreate_data(self):
from devilry.apps.core.models import AssignmentGroup, Candidate, Examiner
from devilry.devilry_group.models import FeedbackSet, ImageAnnotationComment, GroupComment
from devilry.devilry_comment.models import CommentFile
log.info("Rebuilding AssignmentGroup cached data")
log.info("AssignmentGroup count: %s" % AssignmentGroup.objects.count())
log.info("FeedbackSet count: %s" % FeedbackSet.objects.count())
log.info("GroupComments count: %s" % GroupComment.objects.count())
log.info("ImageAnnotationComment count: %s" % ImageAnnotationComment.objects.count())
log.info("CommentFile count: %s" % CommentFile.objects.count())
log.info("Examiner count: %s" % Examiner.objects.count())
log.info("Candidate count: %s" % Candidate.objects.count())
AssignmentGroupCachedData.objects.all().delete()
for period in Period.objects.order_by('-start_time').iterator():
self.execute_sql("""
SELECT devilry__rebuild_assignmentgroupcacheddata_for_period({period_id});
""".format(period_id=period.id))
from devilry.apps.core import models
from devilry.simplified import FieldSpec, FilterSpec, FilterSpecs, ForeignFilterSpec
class SimplifiedCandidateMetaMixin(object):
""" Defines the django model to be used, resultfields returned by
search and which fields can be used to search for a Candidate object
using the Simplified API """
model = models.Candidate
resultfields = FieldSpec('id',
'assignment_group')
searchfields = FieldSpec()
filters = FilterSpecs(FilterSpec('id'),
FilterSpec('assignment_group'),
FilterSpec('assignment_group__parentnode'), # Assignment
FilterSpec('assignment_group__parentnode__parentnode'), # Period
FilterSpec('assignment_group__parentnode__parentnode__parentnode') # Subject
)
def _get_candidate(self, group_id, candidate_id):
try:
return Candidate.objects.get(assignment_group=group_id,
id=candidate_id)
except Candidate.DoesNotExist:
raise NotFoundError(('Candidate with candidate_id={candidate_id} and '
'group_id={group_id} not found').format(group_id=group_id,
candidate_id=candidate_id))
def __get_candidate_queryset(self):
return core_models.Candidate.objects\
.select_related('relatedstudent__user')\
.only(
'candidate_id',
'assignment_group',
'relatedstudent__candidate_id',
'relatedstudent__automatic_anonymous_id',
'relatedstudent__active',
'relatedstudent__user__shortname',
'relatedstudent__user__fullname'
)\
.order_by(
Lower(Concat('relatedstudent__user__fullname',
'relatedstudent__user__shortname')))
for candidate in to_delete:
candidate.delete()
create_kwargs = []
update_candidates = []
for candidatespec in obj.fake_candidates:
username = candidatespec['username']
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
raise InvalidUsername(username)
else:
candidate_id = candidatespec.get('candidate_id', None)
try:
candiate = obj.candidates.get(student__username=username)
except models.Candidate.DoesNotExist:
candidatekwargs = dict(student = user,
candidate_id = candidate_id)
create_kwargs.append(candidatekwargs)
else:
update_candidates.append((candiate, candidate_id))
for candidate, candidate_id in update_candidates:
candidate.candidate_id = candidate_id
candidate.save()
for candidatekwargs in create_kwargs:
obj.candidates.create(**candidatekwargs)
parentnode__parentnode__parentnode=self.assignment.subject,
parentnode__grading_system_plugin_id__in=self.SUPPORTED_GRADING_PLUGINS,
parentnode__parentnode__start_time__gte=self.from_period.start_time,
parentnode__short_name=self.assignment.short_name,
parentnode__parentnode__start_time__lt=self.assignment.parentnode.start_time,
cached_data__last_published_feedbackset__in=feedbackset_queryset
).select_related('parentnode__parentnode', 'cached_data__last_published_feedbackset')
students_on_current = Candidate.objects.filter(
assignment_group__parentnode__parentnode__parentnode=self.assignment.subject,
assignment_group__parentnode=self.assignment,
assignment_group__cached_data__last_published_feedbackset__isnull=True
).select_related('assignment_group__parentnode', 'relatedstudent__user')\
.values_list('relatedstudent__user', flat=True).distinct()
candidates = Candidate.objects.filter(
assignment_group__parentnode__parentnode__parentnode=self.assignment.subject,
assignment_group__in=group_queryset,
relatedstudent__user__in=students_on_current
).select_related('relatedstudent__user',
'assignment_group__parentnode__parentnode',
'assignment_group__cached_data')\
.order_by('relatedstudent__user', '-assignment_group__parentnode__publishing_time')\
.distinct('relatedstudent__user')
return candidates