4147136751

Build Type

github

Committed by GitHub

Commit Message

Merge f4e8d5f13 into 58b7dd6e7

Pull Request Pull Request #754: Develop

Run Details

128 of 128 new or added lines in 6 files covered. (100.0%)

4322 of 7781 relevant lines covered (55.55%)

0.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.67

/sis_provisioner/models/course.py

# Copyright 2023 UW-IT, University of Washington
# SPDX-License-Identifier: Apache-2.0


from django.db import models
from django.conf import settings
from django.utils.timezone import utc, localtime
from sis_provisioner.models import Import, ImportResource
from sis_provisioner.models.group import Group
from sis_provisioner.models.user import User
from sis_provisioner.models.term import Term
from sis_provisioner.dao.course import (
    valid_canvas_section, get_new_sections_by_term)
from sis_provisioner.dao.canvas import get_active_courses_for_term
from sis_provisioner.exceptions import (
    CoursePolicyException, EmptyQueueException)
from datetime import datetime, timedelta


class CourseManager(models.Manager):
    def find_course(self, canvas_course_id, sis_course_id):
        try:
            course = Course.objects.get(canvas_course_id=canvas_course_id)
        except Course.DoesNotExist:
            if not sis_course_id:
                raise
            course = Course.objects.get(course_id=sis_course_id)
        return course

    def get_linked_course_ids(self, course_id):
        return super(CourseManager, self).get_queryset().filter(
            primary_id=course_id).values_list('course_id', flat=True)

    def get_joint_course_ids(self, course_id):
        return super(CourseManager, self).get_queryset().filter(
            xlist_id=course_id).exclude(course_id=course_id).values_list(
                'course_id', flat=True)

    def queue_by_priority(self, priority=ImportResource.PRIORITY_DEFAULT):
        if priority > Course.PRIORITY_DEFAULT:
            filter_limit = settings.SIS_IMPORT_LIMIT['course']['high']
        else:
            filter_limit = settings.SIS_IMPORT_LIMIT['course']['default']

        pks = super(CourseManager, self).get_queryset().filter(
            priority=priority, course_type=Course.SDB_TYPE,
            queue_id__isnull=True, provisioned_error__isnull=True
        ).order_by(
            'provisioned_date', 'added_date'
        ).values_list('pk', flat=True)[:filter_limit]

        if not len(pks):
            raise EmptyQueueException()

        imp = Import(priority=priority, csv_type='course')
        imp.save()

        super(CourseManager, self).get_queryset().filter(
            pk__in=list(pks)).update(queue_id=imp.pk)

        return imp

    def queued(self, queue_id):
        return super(CourseManager, self).get_queryset().filter(
            queue_id=queue_id)

    def dequeue(self, sis_import):
        User.objects.dequeue(sis_import)

        kwargs = {'queue_id': None}
        if sis_import.is_imported():
            kwargs['provisioned_date'] = sis_import.monitor_date
            kwargs['priority'] = Course.PRIORITY_DEFAULT

        self.queued(sis_import.pk).update(**kwargs)

    def add_to_queue(self, section, queue_id):
        if section.is_primary_section:
            course_id = section.canvas_course_sis_id()
        else:
            course_id = section.canvas_section_sis_id()

        try:
            course = Course.objects.get(course_id=course_id)

        except Course.DoesNotExist:
            if section.is_primary_section:
                primary_id = None
            else:
                primary_id = section.canvas_course_sis_id()

            course = Course(course_id=course_id,
                            course_type=Course.SDB_TYPE,
                            term_id=section.term.canvas_sis_id(),
                            primary_id=primary_id)

        course.queue_id = queue_id
        course.save()
        return course

    def remove_from_queue(self, course_id, error=None):
        try:
            course = Course.objects.get(course_id=course_id)
            course.queue_id = None
            if error is not None:
                course.provisioned_error = True
                course.provisioned_status = error
            course.save()

        except Course.DoesNotExist:
            pass

    def update_status(self, section):
        if section.is_primary_section:
            course_id = section.canvas_course_sis_id()
        else:
            course_id = section.canvas_section_sis_id()

        try:
            course = Course.objects.get(course_id=course_id)
            try:
                valid_canvas_section(section)
                course.provisioned_status = None

            except CoursePolicyException as err:
                course.provisioned_status = 'Primary LMS: {} ({})'.format(
                    section.primary_lms, err)

            if section.is_withdrawn():
                course.priority = Course.PRIORITY_NONE

            course.save()
        except Course.DoesNotExist:
            pass

    def add_all_courses_for_term(self, term):
        term_id = term.canvas_sis_id()
        existing_course_ids = dict((c, p) for c, p in (
            super(CourseManager, self).get_queryset().filter(
                term_id=term_id, course_type=Course.SDB_TYPE
            ).values_list('course_id', 'priority')))

        last_search_date = datetime.utcnow().replace(tzinfo=utc)
        try:
            delta = Term.objects.get(term_id=term_id)
        except Term.DoesNotExist:
            delta = Term(term_id=term_id)

        if delta.last_course_search_date is None:
            delta.courses_changed_since_date = datetime.fromtimestamp(0, utc)
        else:
            delta.courses_changed_since_date = (
                delta.last_course_search_date - timedelta(days=1))

        new_courses = []
        for section_data in get_new_sections_by_term(
                localtime(delta.courses_changed_since_date).date(), term,
                existing=existing_course_ids):

            course_id = section_data['course_id']
            if course_id in existing_course_ids:
                if existing_course_ids[course_id] == Course.PRIORITY_NONE:
                    super(CourseManager, self).get_queryset().filter(
                        course_id=course_id).update(
                            priority=Course.PRIORITY_HIGH)
                continue

            new_courses.append(Course(course_id=course_id,
                                      course_type=Course.SDB_TYPE,
                                      term_id=term_id,
                                      primary_id=section_data['primary_id'],
                                      priority=Course.PRIORITY_HIGH))

        Course.objects.bulk_create(new_courses)

        delta.last_course_search_date = last_search_date
        delta.save()

    def prioritize_active_courses_for_term(self, term):
        for sis_course_id in get_active_courses_for_term(term):
            try:
                course = Course.objects.get(course_id=sis_course_id)
                course.priority = Course.PRIORITY_HIGH
                course.save()
            except Course.DoesNotExist:
                pass

    def deprioritize_all_courses_for_term(self, term):
        super(CourseManager, self).get_queryset().filter(
            term_id=term.canvas_sis_id()).update(priority=Course.PRIORITY_NONE)


class Course(ImportResource):
    """ Represents the provisioned state of a course.
    """
    SDB_TYPE = 'sdb'
    ADHOC_TYPE = 'adhoc'
    TYPE_CHOICES = ((SDB_TYPE, 'SDB'), (ADHOC_TYPE, 'Ad Hoc'))
    RETENTION_EXPIRE_MONTH = 12
    RETENTION_EXPIRE_DAY = 1
    RETENTION_LIFE_SPAN = 5

    course_id = models.CharField(max_length=80, null=True)  # sis_course_id
    canvas_course_id = models.CharField(max_length=10, null=True)
    course_type = models.CharField(max_length=16, choices=TYPE_CHOICES)
    term_id = models.CharField(max_length=30, db_index=True)
    primary_id = models.CharField(max_length=80, null=True)
    xlist_id = models.CharField(max_length=80, null=True)
    added_date = models.DateTimeField(auto_now_add=True)
    created_date = models.DateTimeField(null=True)
    provisioned_date = models.DateTimeField(null=True)
    provisioned_error = models.BooleanField(null=True)
    provisioned_status = models.CharField(max_length=512, null=True)
    expiration_date = models.DateTimeField(null=True)
    expiration_exc_granted_date = models.DateTimeField(null=True)
    expiration_exc_granted_by = models.ForeignKey(User, null=True,
                                                  on_delete=models.SET_NULL)
    expiration_exc_desc = models.CharField(max_length=200, null=True)
    deleted_date = models.DateTimeField(null=True)
    priority = models.SmallIntegerField(
        default=ImportResource.PRIORITY_DEFAULT,
        choices=ImportResource.PRIORITY_CHOICES)
    queue_id = models.CharField(max_length=30, null=True)

    objects = CourseManager()

    def is_sdb(self):
        return self.course_type == self.SDB_TYPE

    def is_adhoc(self):
        return self.course_type == self.ADHOC_TYPE

    def sws_url(self):
        try:
            (year, quarter, curr_abbr, course_num,
                section_id) = self.course_id.split('-', 4)
            sws_url = (
                "/restclients/view/sws/student/v5/course/{year},{quarter},"
                "{curr_abbr},{course_num}/{section_id}.json").format(
                    year=year, quarter=quarter, curr_abbr=curr_abbr,
                    course_num=course_num, section_id=section_id)
        except ValueError:
            sws_url = None

        return sws_url

    def update_priority(self, priority):
        for key, val in self.PRIORITY_CHOICES:
            if val == priority:
                self.priority = key
                self.save()
                return

        raise CoursePolicyException("Invalid priority: '{}'".format(priority))

    @property
    def default_expiration_date(self):
        now = datetime.now()
        expiration = datetime(
            now.year + self.RETENTION_LIFE_SPAN, self.RETENTION_EXPIRE_MONTH,
            self.RETENTION_EXPIRE_DAY, 12).replace(tzinfo=utc)
        try:
            (year, quarter, c, n, s) = self.course_id.split('-')
            year = int(year) + self.RETENTION_LIFE_SPAN + (1 if (
                quarter.lower() in ['summer', 'autumn']) else 0)
            return expiration.replace(year=year)
        except ValueError:
            return expiration.replace(
                year=self.created_date.year + self.RETENTION_LIFE_SPAN) if (
                    self.created_date) else expiration

    def json_data(self, include_sws_url=False):
        try:
            group_models = Group.objects.filter(course_id=self.course_id,
                                                is_deleted__isnull=True)
            groups = list(group_models.values_list("group_id", flat=True))
        except Group.DoesNotExist:
            groups = []

        return {
            "course_id": self.course_id,
            "canvas_course_id": self.canvas_course_id,
            "term_id": self.term_id,
            "xlist_id": self.xlist_id,
            "is_sdb_type": self.is_sdb(),
            "added_date": localtime(self.added_date).isoformat() if (
                self.added_date is not None) else None,
            "provisioned_date": localtime(
                self.provisioned_date).isoformat() if (
                    self.provisioned_date is not None) else None,
            "created_date": localtime(self.created_date).isoformat() if (
                self.created_date is not None) else None,
            "deleted_date": localtime(self.deleted_date).isoformat() if (
                self.deleted_date is not None) else None,
            "expiration_date": localtime(self.expiration_date).isoformat() if (
                self.expiration_date is not None) else None,
            "expiration_exc_granted_date": localtime(
                self.expiration_exc_granted_date).isoformat() if (
                    self.expiration_exc_granted_date is not None) else None,
            "expiration_exc_granted_by": (
                self.expiration_exc_granted_by.net_id if (
                    self.expiration_exc_granted_by is not None) else None),
            "expiration_exc_desc": self.expiration_exc_desc,
            "priority": self.PRIORITY_CHOICES[self.priority][1],
            "provisioned_error": self.provisioned_error,
            "provisioned_status": self.provisioned_status,
            "queue_id": self.queue_id,
            "groups": groups,
            "sws_url": self.sws_url() if (
                include_sws_url and self.is_sdb()) else None,
        }

1	# Copyright 2023 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	from django.db import models	1✔
6	from django.conf import settings	1✔
7	from django.utils.timezone import utc, localtime	1✔
8	from sis_provisioner.models import Import, ImportResource	1✔
9	from sis_provisioner.models.group import Group	1✔
10	from sis_provisioner.models.user import User	1✔
11	from sis_provisioner.models.term import Term	1✔
12	from sis_provisioner.dao.course import (	1✔
13	valid_canvas_section, get_new_sections_by_term)
14	from sis_provisioner.dao.canvas import get_active_courses_for_term	1✔
15	from sis_provisioner.exceptions import (	1✔
16	CoursePolicyException, EmptyQueueException)
17	from datetime import datetime, timedelta	1✔
18
19
20	class CourseManager(models.Manager):	1✔
21	def find_course(self, canvas_course_id, sis_course_id):	1✔
22	try:	×
23	course = Course.objects.get(canvas_course_id=canvas_course_id)	×
24	except Course.DoesNotExist:	×
25	if not sis_course_id:	×
26	raise	×
27	course = Course.objects.get(course_id=sis_course_id)	×
28	return course	×
29
30	def get_linked_course_ids(self, course_id):	1✔
31	return super(CourseManager, self).get_queryset().filter(	×
32	primary_id=course_id).values_list('course_id', flat=True)
33
34	def get_joint_course_ids(self, course_id):	1✔
35	return super(CourseManager, self).get_queryset().filter(	×
36	xlist_id=course_id).exclude(course_id=course_id).values_list(
37	'course_id', flat=True)
38
39	def queue_by_priority(self, priority=ImportResource.PRIORITY_DEFAULT):	1✔
40	if priority > Course.PRIORITY_DEFAULT:	×
41	filter_limit = settings.SIS_IMPORT_LIMIT['course']['high']	×
42	else:
43	filter_limit = settings.SIS_IMPORT_LIMIT['course']['default']	×
44
45	pks = super(CourseManager, self).get_queryset().filter(	×
46	priority=priority, course_type=Course.SDB_TYPE,
47	queue_id__isnull=True, provisioned_error__isnull=True
48	).order_by(
49	'provisioned_date', 'added_date'
50	).values_list('pk', flat=True)[:filter_limit]
51
52	if not len(pks):	×
53	raise EmptyQueueException()	×
54
55	imp = Import(priority=priority, csv_type='course')	×
56	imp.save()	×
57
58	super(CourseManager, self).get_queryset().filter(	×
59	pk__in=list(pks)).update(queue_id=imp.pk)
60
61	return imp	×
62
63	def queued(self, queue_id):	1✔
64	return super(CourseManager, self).get_queryset().filter(	1✔
65	queue_id=queue_id)
66
67	def dequeue(self, sis_import):	1✔
68	User.objects.dequeue(sis_import)	1✔
69
70	kwargs = {'queue_id': None}	1✔
71	if sis_import.is_imported():	1✔
72	kwargs['provisioned_date'] = sis_import.monitor_date	1✔
73	kwargs['priority'] = Course.PRIORITY_DEFAULT	1✔
74
75	self.queued(sis_import.pk).update(**kwargs)	1✔
76
77	def add_to_queue(self, section, queue_id):	1✔
78	if section.is_primary_section:	1✔
79	course_id = section.canvas_course_sis_id()	1✔
80	else:
81	course_id = section.canvas_section_sis_id()	1✔
82
83	try:	1✔
84	course = Course.objects.get(course_id=course_id)	1✔
85
86	except Course.DoesNotExist:	1✔
87	if section.is_primary_section:	1✔
88	primary_id = None	1✔
89	else:
90	primary_id = section.canvas_course_sis_id()	1✔
91
92	course = Course(course_id=course_id,	1✔
93	course_type=Course.SDB_TYPE,
94	term_id=section.term.canvas_sis_id(),
95	primary_id=primary_id)
96
97	course.queue_id = queue_id	1✔
98	course.save()	1✔
99	return course	1✔
100
101	def remove_from_queue(self, course_id, error=None):	1✔
102	try:	1✔
103	course = Course.objects.get(course_id=course_id)	1✔
104	course.queue_id = None	1✔
105	if error is not None:	1✔
106	course.provisioned_error = True	1✔
107	course.provisioned_status = error	1✔
108	course.save()	1✔
109
110	except Course.DoesNotExist:	1✔
111	pass	1✔
112
113	def update_status(self, section):	1✔
114	if section.is_primary_section:	1✔
115	course_id = section.canvas_course_sis_id()	1✔
116	else:
117	course_id = section.canvas_section_sis_id()	×
118
119	try:	1✔
120	course = Course.objects.get(course_id=course_id)	1✔
121	try:	1✔
122	valid_canvas_section(section)	1✔
123	course.provisioned_status = None	1✔
124
125	except CoursePolicyException as err:	×
126	course.provisioned_status = 'Primary LMS: {} ({})'.format(	×
127	section.primary_lms, err)
128
129	if section.is_withdrawn():	1✔
130	course.priority = Course.PRIORITY_NONE	1✔
131
132	course.save()	1✔
133	except Course.DoesNotExist:	×
134	pass	×
135
136	def add_all_courses_for_term(self, term):	1✔
137	term_id = term.canvas_sis_id()	×
138	existing_course_ids = dict((c, p) for c, p in (	×
139	super(CourseManager, self).get_queryset().filter(
140	term_id=term_id, course_type=Course.SDB_TYPE
141	).values_list('course_id', 'priority')))
142
143	last_search_date = datetime.utcnow().replace(tzinfo=utc)	×
144	try:	×
145	delta = Term.objects.get(term_id=term_id)	×
146	except Term.DoesNotExist:	×
147	delta = Term(term_id=term_id)	×
148
149	if delta.last_course_search_date is None:	×
150	delta.courses_changed_since_date = datetime.fromtimestamp(0, utc)	×
151	else:
152	delta.courses_changed_since_date = (	×
153	delta.last_course_search_date - timedelta(days=1))
154
155	new_courses = []	×
156	for section_data in get_new_sections_by_term(	×
157	localtime(delta.courses_changed_since_date).date(), term,
158	existing=existing_course_ids):
159
160	course_id = section_data['course_id']	×
161	if course_id in existing_course_ids:	×
162	if existing_course_ids[course_id] == Course.PRIORITY_NONE:	×
163	super(CourseManager, self).get_queryset().filter(	×
164	course_id=course_id).update(
165	priority=Course.PRIORITY_HIGH)
166	continue	×
167
168	new_courses.append(Course(course_id=course_id,	×
169	course_type=Course.SDB_TYPE,
170	term_id=term_id,
171	primary_id=section_data['primary_id'],
172	priority=Course.PRIORITY_HIGH))
173
174	Course.objects.bulk_create(new_courses)	×
175
176	delta.last_course_search_date = last_search_date	×
177	delta.save()	×
178
179	def prioritize_active_courses_for_term(self, term):	1✔
180	for sis_course_id in get_active_courses_for_term(term):	×
181	try:	×
182	course = Course.objects.get(course_id=sis_course_id)	×
183	course.priority = Course.PRIORITY_HIGH	×
184	course.save()	×
185	except Course.DoesNotExist:	×
186	pass	×
187
188	def deprioritize_all_courses_for_term(self, term):	1✔
189	super(CourseManager, self).get_queryset().filter(	×
190	term_id=term.canvas_sis_id()).update(priority=Course.PRIORITY_NONE)
191
192
193	class Course(ImportResource):	1✔
194	""" Represents the provisioned state of a course.
195	"""
196	SDB_TYPE = 'sdb'	1✔
197	ADHOC_TYPE = 'adhoc'	1✔
198	TYPE_CHOICES = ((SDB_TYPE, 'SDB'), (ADHOC_TYPE, 'Ad Hoc'))	1✔
199	RETENTION_EXPIRE_MONTH = 12	1✔
200	RETENTION_EXPIRE_DAY = 1	1✔
201	RETENTION_LIFE_SPAN = 5	1✔
202
203	course_id = models.CharField(max_length=80, null=True) # sis_course_id	1✔
204	canvas_course_id = models.CharField(max_length=10, null=True)	1✔
205	course_type = models.CharField(max_length=16, choices=TYPE_CHOICES)	1✔
206	term_id = models.CharField(max_length=30, db_index=True)	1✔
207	primary_id = models.CharField(max_length=80, null=True)	1✔
208	xlist_id = models.CharField(max_length=80, null=True)	1✔
209	added_date = models.DateTimeField(auto_now_add=True)	1✔
210	created_date = models.DateTimeField(null=True)	1✔
211	provisioned_date = models.DateTimeField(null=True)	1✔
212	provisioned_error = models.BooleanField(null=True)	1✔
213	provisioned_status = models.CharField(max_length=512, null=True)	1✔
214	expiration_date = models.DateTimeField(null=True)	1✔
215	expiration_exc_granted_date = models.DateTimeField(null=True)	1✔
216	expiration_exc_granted_by = models.ForeignKey(User, null=True,	1✔
217	on_delete=models.SET_NULL)
218	expiration_exc_desc = models.CharField(max_length=200, null=True)	1✔
219	deleted_date = models.DateTimeField(null=True)	1✔
220	priority = models.SmallIntegerField(	1✔
221	default=ImportResource.PRIORITY_DEFAULT,
222	choices=ImportResource.PRIORITY_CHOICES)
223	queue_id = models.CharField(max_length=30, null=True)	1✔
224
225	objects = CourseManager()	1✔
226
227	def is_sdb(self):	1✔
228	return self.course_type == self.SDB_TYPE	1✔
229
230	def is_adhoc(self):	1✔
231	return self.course_type == self.ADHOC_TYPE	1✔
232
233	def sws_url(self):	1✔
234	try:	1✔
235	(year, quarter, curr_abbr, course_num,	1✔
236	section_id) = self.course_id.split('-', 4)
237	sws_url = (	1✔
238	"/restclients/view/sws/student/v5/course/{year},{quarter},"
239	"{curr_abbr},{course_num}/{section_id}.json").format(
240	year=year, quarter=quarter, curr_abbr=curr_abbr,
241	course_num=course_num, section_id=section_id)
242	except ValueError:	1✔
243	sws_url = None	1✔
244
245	return sws_url	1✔
246
247	def update_priority(self, priority):	1✔
248	for key, val in self.PRIORITY_CHOICES:	1✔
249	if val == priority:	1✔
250	self.priority = key	1✔
251	self.save()	1✔
252	return	1✔
253
254	raise CoursePolicyException("Invalid priority: '{}'".format(priority))	1✔
255
256	@property	1✔
257	def default_expiration_date(self):	1✔
258	now = datetime.now()	1✔
259	expiration = datetime(	1✔
260	now.year + self.RETENTION_LIFE_SPAN, self.RETENTION_EXPIRE_MONTH,
261	self.RETENTION_EXPIRE_DAY, 12).replace(tzinfo=utc)
262	try:	1✔
263	(year, quarter, c, n, s) = self.course_id.split('-')	1✔
264	year = int(year) + self.RETENTION_LIFE_SPAN + (1 if (	1✔
265	quarter.lower() in ['summer', 'autumn']) else 0)
266	return expiration.replace(year=year)	1✔
267	except ValueError:	1✔
268	return expiration.replace(	1✔
269	year=self.created_date.year + self.RETENTION_LIFE_SPAN) if (
270	self.created_date) else expiration
271
272	def json_data(self, include_sws_url=False):	1✔
273	try:	×
274	group_models = Group.objects.filter(course_id=self.course_id,	×
275	is_deleted__isnull=True)
276	groups = list(group_models.values_list("group_id", flat=True))	×
277	except Group.DoesNotExist:	×
278	groups = []	×
279
280	return {	×
281	"course_id": self.course_id,
282	"canvas_course_id": self.canvas_course_id,
283	"term_id": self.term_id,
284	"xlist_id": self.xlist_id,
285	"is_sdb_type": self.is_sdb(),
286	"added_date": localtime(self.added_date).isoformat() if (
287	self.added_date is not None) else None,
288	"provisioned_date": localtime(
289	self.provisioned_date).isoformat() if (
290	self.provisioned_date is not None) else None,
291	"created_date": localtime(self.created_date).isoformat() if (
292	self.created_date is not None) else None,
293	"deleted_date": localtime(self.deleted_date).isoformat() if (
294	self.deleted_date is not None) else None,
295	"expiration_date": localtime(self.expiration_date).isoformat() if (
296	self.expiration_date is not None) else None,
297	"expiration_exc_granted_date": localtime(
298	self.expiration_exc_granted_date).isoformat() if (
299	self.expiration_exc_granted_date is not None) else None,
300	"expiration_exc_granted_by": (
301	self.expiration_exc_granted_by.net_id if (
302	self.expiration_exc_granted_by is not None) else None),
303	"expiration_exc_desc": self.expiration_exc_desc,
304	"priority": self.PRIORITY_CHOICES[self.priority][1],
305	"provisioned_error": self.provisioned_error,
306	"provisioned_status": self.provisioned_status,
307	"queue_id": self.queue_id,
308	"groups": groups,
309	"sws_url": self.sws_url() if (
310	include_sws_url and self.is_sdb()) else None,
311	}

uw-it-aca / canvas-sis-provisioner / 4147136751

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous