Commit 16579d44 authored by Kevin Lin's avatar Kevin Lin
Browse files

Initial commit

parents
import collections
import json
import re
import requests
class Canvas(object):
def __init__(self, defaultAuthToken=None, defaultServer="canvas.uw.edu", defaultVersion='v1', defaultPerPage=10000):
self.defaultServer = defaultServer
self.defaultAuthToken = defaultAuthToken
self.defaultVersion = defaultVersion
self.defaultPerPage = defaultPerPage
# set API access clases to None
self.courses = None
def pages(self, url, query, absoluteUrl=False, verbose=False):
# Make the initial call to the API to retrieve the first page
# of results.
resp = self.callAPI(url, query, absoluteUrl, verbose)
# Always send back the first page of results.
yield resp
checkMorePages = True
while checkMorePages:
# check for 'Link' header
linkHdr = None
for h in resp.headers.items():
if h[0] == 'link':
linkHdr = h
break
else:
# If we didn't find a 'Link' header, stop.
checkMorePages = False
return
# If we DID find a 'Link' header, then search for a 'next' link.
# All the links are separated by commas.
links = linkHdr[1].split(',')
for lnk in links:
# Each link consists of two parts (the URL, and the 'rel'
# attribute), separated by a semi-colon.
parts = lnk.split(';')
if parts[1].find('next') >= 0:
nextPageUrl = parts[0]
# Remove the < and > characters that delineate the URL.
nextPageUrl = nextPageUrl.replace('<', '')
nextPageUrl = nextPageUrl.replace('>', '')
nextPageUrl = nextPageUrl.strip()
# Get the URL to retrieve the next page of results.
resp = self.callAPI(nextPageUrl, query, absoluteUrl=True)
# Wait until someone asks for the next page.
yield resp
break
else:
# If we went through all of the links and didn't find a 'next'
# link, then stop looking.
checkMorePages = False
return
def get(self, url, query=None, absoluteUrl=False, verbose=False):
if query is None:
query = {}
# A place to accumulate all of the results from all of the pages.
collector = []
# Read all of the pages of results from this API call.
for pg in self.pages(url, query, absoluteUrl, verbose):
respBody = pg.json()
# If we were calling an API that returns single objects and not
# a list of objects, then wrap the object in a list so that we
# can append it to the collector list.
if isinstance(respBody, dict):
respBody = [respBody]
collector = collector + respBody
return collector
def put(self, url, params, absoluteUrl=False):
# Gather connection/API information
server = self.defaultServer
authToken = self.defaultAuthToken
version = self.defaultVersion
perPage = self.defaultPerPage
if absoluteUrl:
urlstr = url
else:
if url.find('?') < 0:
urlstr = 'https://{}/api/{}/{}?per_page={}'.format(server, version, url, perPage)
else:
urlstr = 'https://{}/api/{}/{}&per_page={}'.format(server, version, url, perPage)
# Create the request, adding in the oauth authorization token
req = requests.put(urlstr, headers={'Authorization':'Bearer {}'.format(authToken)}, params=params)
return req.json()
def delete(self, url, absoluteUrl=False):
# Gather connection/API information
server = self.defaultServer
authToken = self.defaultAuthToken
version = self.defaultVersion
perPage = self.defaultPerPage
if absoluteUrl:
urlstr = url
else:
if url.find('?') < 0:
urlstr = 'https://{}/api/{}/{}?per_page={}'.format(server, version, url, perPage)
else:
urlstr = 'https://{}/api/{}/{}&per_page={}'.format(server, version, url, perPage)
# Create the request, adding in the oauth authorization token
req = requests.delete(urlstr, headers={'Authorization':'Bearer {}'.format(authToken)})
return req.json()
def post(self, url, data, absoluteUrl=False):
# Gather connection/API information
server = self.defaultServer
authToken = self.defaultAuthToken
version = self.defaultVersion
perPage = self.defaultPerPage
if absoluteUrl:
urlstr = url
else:
if url.find('?') < 0:
urlstr = 'https://{}/api/{}/{}?per_page={}'.format(server, version, url, perPage)
else:
urlstr = 'https://{}/api/{}/{}&per_page={}'.format(server, version, url, perPage)
headers = {
'content-type': 'application/json',
'Authorization':'Bearer {}'.format(authToken)
}
# Create the request, adding in the oauth authorization token
req = requests.post(urlstr, headers=headers, data=json.dumps(data))
return req.json()
def callAPI(self, url, query, absoluteUrl=False, verbose=False):
# Gather connection/API information
server = self.defaultServer
if server == None:
raise ValueError('Property \'defaultServer\' must be set prior to calling callAPI.')
authToken = self.defaultAuthToken
if authToken == None:
raise ValueError('Property \'defaultAuthToken\' must be set prior to calling callAPI.')
version = self.defaultVersion
if version == None:
raise ValueError('Property \'defaultVersion\' must be set prior to calling callAPI.')
perPage = self.defaultPerPage
# If we're dealing with an absolute URL, then use it as it
# was provided.
if absoluteUrl:
urlstr = url
# Otherwise, augment the URL with the default server, version, and
# per_page information.
else:
if url.find('?') < 0:
urlstr = 'https://{}/api/{}/{}?per_page={}'.format(server, version, url, perPage)
else:
urlstr = 'https://{}/api/{}/{}&per_page={}'.format(server, version, url, perPage)
# Print to standard out only if verbose == True.
if verbose:
print('Attempting to retrieve {} ...'.format(urlstr))
# Create the request, adding in the oauth authorization token
req = requests.get(urlstr, headers={'Authorization':'Bearer {}'.format(authToken)}, params=query)
return req
def get_course(self, course_id):
course = self.get("/courses/%s" % course_id)[0]
if 'sis_course_id' not in course:
return None
course = course['sis_course_id'].split("-")
name = course[2] + course[3]
quarter = course[1].title() + ' ' + course[0]
return Course(self, course_id, name=name, section=course[4], quarter=quarter)
def get_courses_taught(self):
return self.get_courses("teacher")
def get_courses(self, only=None):
params = {}
if only is not None:
params['enrollment_type'] = only
course_names = []
for course in self.get("/courses", params):
if 'errors' in course:
return None
if 'sis_course_id' in course:
course_names.append((course['sis_course_id'].split("-")[0:5], course['id']))
def sort_key(course_data):
(year, quarter, a, b, c), cid = course_data
quarters = {
'winter': 1,
'spring': 2,
'summer': 3,
'autumn': 4,
}
return int(year), quarters[quarter], a + b + " " + c
course_names.sort(key=sort_key)
courses = []
for (course, course_id) in course_names:
name = course[2] + course[3]
quarter = course[1].title() + ' ' + course[0]
courses.append(Course(self, course_id, name=name, section=course[4], quarter=quarter))
return courses
class Course(object):
def __init__(self, canvas, course_id, name=None, section=None, quarter=None):
self._canvas = canvas
self.id = course_id
self.name = name
self.section = section
self.quarter = quarter
def __str__(self):
return self.name + " " + self.section + " (" + self.quarter + ")"
def __repr__(self):
return str(self)
def json(self):
return {'id': str(self.id), 'name': self.name, 'section': self.section, 'quarter': self.quarter}
def remove_all_tabs(self):
for tab in self._canvas.get("/courses/%s/tabs" % self.id):
self._canvas.put("/courses/%s/tabs/%s/" % (self.id, tab['id']), {"hidden":True})
self._canvas.put("/courses/%s/tabs/grades" % self.id, {"hidden":False})
def write_settings(self):
self._canvas.put("/courses/%s/settings/" % self.id, {
'allow_student_organized_groups': False,
'hide_final_grades': True,
'hide_distribution_graphs': True,
'allow_student_discussion_topics': False,
'allow_student_discussion_editing': False,
'lock_all_announcements': True
})
def get_assignment_group(self, group_id):
group = self._canvas.get('/courses/%s/assignment_groups/%s' % (self.id, group_id),
{"include": ["assignments"]})[0]
return AssignmentGroup(self, group['id'], name=group['name'], x_assignments=group['assignments'])
def get_assignment_groups(self):
if hasattr(self, 'groups'):
return self.groups
groups = self._canvas.get('/courses/%s/assignment_groups' % self.id)
groups.sort(key=lambda x: x['name'])
self.groups = []
for group in groups:
self.groups.append(AssignmentGroup(self, group['id'], name=group['name']))
return self.groups
def get_groups(self):
if hasattr(self, 'groups'):
return self.groups
self.groups = {}
result = self._canvas.get('/courses/%s/groups' % self.id)
for group in result:
users = self._canvas.get('/groups/%d/users' % group['id'])
self.groups[group['name']] = [Student(self, u['login_id'], u['id'], u['name']) for u in users]
return self.groups
def get_students(self):
if hasattr(self, 'students'):
return self.students
self.students = []
self.students_map = {}
result = self._canvas.get('/courses/%s/students' % self.id)
for student in result:
student = Student(self, student['login_id'], student['id'], student['name'])
self.students.append(student)
self.students_map[student.uwnetid] = student
return self.students
def get_students_map(self):
self.get_students()
return self.students_map
def create_assignment(self, assn_group, name, points, description=""):
assignment = self._canvas.post("/courses/%s/assignments" % self.id, {'assignment':{
"name": name,
"assignment_group_id": assn_group.id,
"submission_types": ["none"],
"notify_of_update": False,
"points_possible": points,
"description": description,
"muted": True,
"published": True
}})
return Assignment(self, assignment['id'], name, points, description, assn_group, assignment['muted'], True)
def get_assignment(self, assn_id):
try:
assn = self._canvas.get('courses/%s/assignments/%s' % (self.id, assn_id))[0]
automatic = len(assn['submission_types']) == 1 and assn['submission_types'][0] == 'none'
return Assignment(self, assn['id'], assn['name'], assn['points_possible'], assn['description'], AssignmentGroup(self, assn['assignment_group_id']), assn['muted'], automatic)
except:
return None
class AssignmentGroup(object):
def __init__(self, course, group_id, name=None, x_assignments=None):
self.course = course
self.id = group_id
self.name = name
self.x_assignments = x_assignments
def __str__(self):
return self.name or "??? (id=" + str(self.id) + ")"
def __repr__(self):
return str(self)
def create_assignment(self, name, due_date, lock_date, points, online, description=""):
settings = {
"name": name,
"assignment_group_id": self.id,
"submission_types": ["none" if online else "online_upload"],
"notify_of_update": False,
"points_possible": points,
"description": description,
"muted": True,
"published": True,
"due_at": due_date.isoformat(),
"lock_at": lock_date.isoformat()
}
if not online:
settings["allowed_extensions"] = ["pdf"]
assignment = self.course._canvas.post("/courses/%s/assignments" % self.course.id, {'assignment': settings})
return Assignment(self, self.course, assignment['id'], name, points, description, self, assignment['muted'], True)
def json(self):
return {'id': str(self.id), 'name': self.name, 'course': self.course.id}
def get_assignments(self):
if hasattr(self, 'assignments'):
return self.assignments
if self.x_assignments:
assns = self.x_assignments
else:
assns = self.course._canvas.get('/courses/%s/assignment_groups/%s' % (self.course.id, self.id), {"include": ["assignments"]})[0]["assignments"]
assns.sort(key=lambda x: x['name'])
self.assignments = []
for assn in assns:
automatic = len(assn['submission_types']) == 1 and assn['submission_types'][0] == 'none'
self.assignments.append(Assignment(self.course, assn['id'], assn['name'], assn['points_possible'], assn['description'], AssignmentGroup(self, assn['group_category_id']), assn['muted'], automatic))
return self.assignments
class Student(object):
def __init__(self, course, uwnetid, student_id, name=None):
self._canvas = course._canvas
self.course = course
self.uwnetid = uwnetid
self.id = student_id
self.name = name
def __str__(self):
return self.name + " (" + self.uwnetid + ")"
def __repr__(self):
return str(self)
class Assignment(object):
def __init__(self, course, assn_id, name=None, points=None, description="", group=None, muted=False, automatic=False, canvas=None):
if canvas is not None:
self._canvas = canvas
else:
self._canvas = course._canvas
self.course = course if hasattr(course, 'id') else None
self.course_id = course.id if hasattr(course, 'id') else course
self.id = assn_id
self.name = name
self.points = points
self.description = description
self.group = group
self.muted = muted
self.automatic = automatic
def __str__(self):
return self.name or "??? (id=" + str(self.id) + ")"
def __repr__(self):
return str(self)
def set_grades(self, scores, text=None):
if text is None:
text = "grinch.cs has auto-submitted a score"
canvas_scores = {}
for score in scores:
canvas_scores[(score[0].id if hasattr(score[0], 'id') else score[0])] = {'posted_grade': score[1], 'text_comment': score[2] if len(score) > 2 else text}
self._canvas.post("/courses/%s/assignments/%s/submissions/update_grades" % (self.course_id, self.id), {'grade_data': canvas_scores})
def edit(self, name=None, points=None, muted=True, description=None):
if name is None:
name = self.name
if points is None:
points = self.points
if description is None:
description = self.description
assignment = self._canvas.put("/courses/%s/assignments/%s" % (self.course.id, self.id), {'assignment':{
"id": self.id,
"name": name,
"points_possible": points,
"description": description,
"muted": muted
}})
self.name = assignment['name']
self.points = assignment['points']
self.description = assignment['description']
self.muted = assignment['muted']
return self
def get_overrides(self):
return self.course._canvas.get('/courses/%s/assignments/%s/overrides' % (self.course.id, self.id))
def edit_override(self, override):
self._canvas.delete("courses/%s/assignments/%s/overrides/%s.json" % (self.course.id, self.id, override['id']))
self._canvas.post("courses/%s/assignments/%s/overrides.json" % (self.course.id, self.id), {'assignment_override':{
'student_ids': override['student_ids'],
'title': override['title'],
'due_at': None,
'lock_at': override['lock_at']
}})
def override(self, name, student, due):
self._canvas.post("courses/%s/assignments/%s/overrides.json" % (self.course.id, self.id), {'assignment_override':{
'student_ids': [student.id],
'title': name + " for " + student.uwnetid,
'due_at': None,
'lock_at': due.isoformat()
}})
#!/usr/bin/env python3
import csv, os
import canvas_api
import os.path
import readline
from config import token_file, course_id
if __name__ == "__main__":
if not os.path.isfile(token_file):
token = input("Enter a Canvas Token (https://canvas.uw.edu/profile/settings): ")
with open(token_file, 'w') as f:
f.write(token)
token = open(token_file, 'r').read().strip(" \n\t")
canvas = canvas_api.Canvas(token)
courses_taught = canvas.get_courses_taught()
if courses_taught is None:
print("Invalid Canvas token.")
exit(1)
course = canvas.get_course(course_id)
if course is None:
print("Invalid Canvas course.")
exit(1)
groups = course.get_groups()
print("Pre-assign Room Name,Email Address")
for group in groups:
if not groups[group]:
print(group + ",test@uw.edu")
continue
for student in groups[group]:
print(group + "," + student.uwnetid + "@uw.edu")
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment