This blog describes how to move user enrollments from one role to another using a Python class, SQL data, and a mapping file.
So here is the situation we are presently facing at Everett Public Schools. Along with our base roles of Student, Teacher, Designer, etc., we also have custom roles that have been derived from those base roles. These custom roles are a bit more refined and help keep users and there permissions in check. The problem with this idea is that not everyone follows the rules when assigning a role to a user when that user is enrolled into a course. This quickly becomes an issue when trying to search and sort users based upon their permissions.
Case in point: We have teachers that are enrolled as students in staff courses or portals that are located at their respective school or sub-account. So are they truly a student in the classic sense? No. When you do a blind search for students, you get back a bunch of teachers and maybe a few other users that somebody down the line added to a course as a student. Now that the user data set has gotten out of hand, how do you move those enrollments over to the new custom role that you just created? In addition to that, how do you keep it all in sync?
The solution comes in a few simple steps which you can follow below. First, you need to decide what data set of users need to be moved from one role to another? In our case, we wanted non-students (i.e. district staff) that were currently assigned the base role of StudentEnrollment (aka Student). These district IDs are the same as their login id and SIS id too, so it keeps things straight. Since we run multiple nightly integrations, we simply just created a new section in our SQL code to only pull the district staff IDs. Like this:
/*
STAFF USERS
*/
IF @type = 'STAFF_USERS'
BEGIN
SELECT login_id
FROM eps_canvas.dbo.users
WHERE user_type = 'F';
END
Just a bit of a backstory to explain the logic. In Everett we use several nightly imports into Canvas to roster courses, control users, etc. More on that in another blog, but to suffice it to say it works very well. We use a 'users' table in a smaller database to control who gets put into Canvas. The user_type of 'F' is for 'faculty'. So when this script runs, it uses the 'staff_users' input parameter to control what data set the script will receive. This logic comes from the script configuration .ini file:
[Default]
#API SIS upload URL for the site
#Root account should always be 1
rootURL: https://everettsd.beta.instructure.com/api/v1/#The URL string data that allows acting as another user
#The 'replace' placeholder gets replaced with the correct term in the script
masqueradeData: {"as_user_id": "replace"}#The list of parameters to pull from the DB
#Use this list to effect the role mappping below
#Comma delimited, any order
dbParams: staff_users#Text of the SQL Server stored procedure SQL
#For getting of district ids
dbSQL: exec eps_internal.dbo.pyCurGetCanvasCustomExtracts ?#The endpoint to get enrollments for a user
enrollmentsEndpoint: users/self/enrollments#The endpoint to enroll the user in the course
coursesEndpoint: courses/{}/enrollments#The endpoint to get all of the current roles
rolesEndpoint: accounts/1/roles#The mapping from one role to another for each DB parameter
#The key for each map is keyed off of the dbParams list
#The JSON object for each dbParam is a key of the permission type to find, the value is the role to assign
#All values are case sensitive and must match exactly to what is in Canvas
roleMapping: {"staff_users": {"StudentEnrollment": "Adult Learner"}}
When the script is executed, it looks for an associated configuration file and reads in the [Default] section data. It does read a master configuration file too so it can set some global variables, but that is outside the scope of this post. Each parameter is then assigned to an internal variable that the script uses to do its thing. Jumping down to the bottom line in the file, the roleMapping dictionary is keyed to the dbParams value. This is how the data set knows what users to process, what role to look for (in this case 'StudentEnrollment') and what role to use when enrolling the user into the current course ('Adult Learner'). If we wanted to process more users this this script workflow, then we add a value to the dbParams list and add the same value to the roleMapping dictionary along with the roles to use.
At some point, we needed to create our 'Adult Learner' role. We wanted a role that was student based but that could be used for staff members that are fulfilling some student role in a course somewhere. We wanted the student role to truly reflect actual students in the district.
So now we are ready to roll. Consider this Python class:
from requests import Session
from classEpsDB import EpsDB
from classEpsException import EpsException
from classEpsConfiguration import EpsConfiguration
from json import loads
from urllib import parse
class EpsITSyncCanvasEnrollments(object😞
"""
Syncs the Canvas enrollments between what was assigned to a user and what should be the correct assignment.
We do this to keep users from getting the incorrect enrollment and streamlining the search process.
@package: epsIT
@license: http://opensource.org/licenses/GPL-3.0
@copyright: 2020, Everett Public Schools
@author: DPassey
@version: 1.0, 02.24.2020
"""
def __init__(self, user_id_type='sis_user_id'😞
"""
Class initializer.
Parses the config file_name, assigning values as needed.
@raise exception: EpsException
"""
try:
cfg = EpsConfiguration(f"{self.__class__.__name__}.ini")
self.rc = 0
if not cfg.db_dsn: raise Exception(f"{self.__class__.__name__}.__init__. DSN data source is missing.")
for k in cfg.locals:
k = k.upper().strip()
v = cfg.locals[k].strip()
if k == 'DBSQL': db_sql = v
if k == 'DBPARAMS': param_list = v.split(',')
if k == 'ROOTURL': root_url = v
if k == 'MASQUERADEDATA': masquerade = v
if k == 'ENROLLMENTSENDPOINT': enroll_endpoint = v
if k == 'COURSESENDPOINT': course_endpoint = v
if k == 'ROLEMAPPING': roles_map = loads(v)
if k == 'ROLESENDPOINT': roles_endpoint = v
# set the session header
self.header = {'Authorization': f'Bearer {cfg.canvas_token}'}
# must be one of these
if user_id_type not in ('sis_user_id', 'sis_login_id'😞 raise Exception(f'{self.__class__.__name__}.__init__. Invalid parameter: {user_id_type}.')
# create a session
with Session() as self.session:
# get the type of user from the parameter list
for _ in param_list:
# get all of the active roles
url = f"{root_url}{roles_endpoint}"
# for each mapped role for this parameter, get the role's id
roles_dict = self.get_account_roles(url, roles_map[_])
# get the data to process for each parameter
data = self.get_data(cfg.db_dsn, db_sql, _)
# proceed if we get user data
if data:
# for each user in the data, find the applicable enrollments to move
for user in data:
# set up masquerading
self.data_dict = loads(masquerade.replace('replace', "{}:{}".format(user_id_type, user[0])))
# get all of the user's enrollments to see if we need to change enrollments
user_dict = self.get_enrollments(f"{root_url}{enroll_endpoint}", roles_map[_])
# now process the users by their Canvas id
for user_id in user_dict:
# process each course and re-enroll the user
# we need to keep the indexing linked between course and enrollment
for c, course in enumerate(user_dict[user_id]['courses']):
# get the role id of the new role
# need this to move enrollments
role_id = roles_dict[user_dict[user_id]['roles'][c]]
# get the current enrollment id
enroll_id = user_dict[user_id]['enrollments'][c]
endpoint = course_endpoint.format(course)
# now set the new enrollments
self.set_enrollment(f"{root_url}{endpoint}", user_id, role_id, enroll_id)
except:
EpsException(__file__)
def get_data(self, dsn, sql, param):
"""
Executes the stored procedure and gets the applicable data set.
@param dsn: String
@param sql: String
@param param: String
@return: List
@raise exception: EpsException
"""
try:
db = EpsDB(dsn)
if not db: raise Exception(f"{self.__class__.__name__}.get_data. Could not connect to database.")
rs = db.get(sql, param)
if not rs: raise Exception(f"{self.__class__.__name__}.get_data. No data set returned.")
return rs
except:
EpsException(__file__)
def get_account_roles(self, url, role_dict):
"""
Gets the active roles and puts them in a roles dictionary.
@param url: String
@param role_dict: Dictionary
@return Dictionary
@raise exception: EpsException
"""
try:
role_id_dict = {}
# get all active roles
data_dict = {'state[]': 'active', 'per_page': 100}
resp = self.session.get(url, data=data_dict, headers=self.header)
if resp.status_code == 200:
# check the headers "link" attribute for the last relational link
for link in resp.headers['Link'].split(','😞
if 'rel=last' in link.replace('"','').replace("'",'').lower():
# grab the total pages count by parsing out the url parts and convert to int
page_total = int(parse.parse_qs(parse.urlparse(link.split(';')[0])[4])['page'][0])
# we need to get all results since we are being paginated
# these sections perform the same logic, just easier to to write it this way
if page_total > 1:
p = 1
while p <= page_total:
data_dict.update({'page': p})
resp = self.session.get(url, data=data_dict, headers=self.header)
json = loads(resp.text)
for _ in json:
if _['role'] in role_dict.values(): role_id_dict[_['role']] = _['id']
p += 1
else:
json = loads(resp.text)
for _ in json:
if _['role'] in role_dict.values(): role_id_dict[_['role']] = _['id']
else: raise Exception(f"{self.__class__.__name__}.get_account_roles. Response {resp.text} returned.")
return role_id_dict
except:
EpsException(__file__)
def get_enrollments(self, url, map_dict):
"""
Gets the roles for the user and place in a user dictionary.
@param url: String
@param map_dict: Dictionary
@return Dictionary
@raise exception: EpsException
"""
try:
user_list = []
enrollments_list = []
roles_list = []
user_dict = {}
# make a copy of the class data dictionary so we can update it
data_dict = self.data_dict.copy()
# we should never exceed the per_page value
# i mean really....over 100 enrollments?
# current_and_future is a special state for all courses, published and unpublished
data_dict.update({'state[]': 'current_and_future', 'per_page': 100})
resp = self.session.get(url, data=data_dict, headers=self.header)
if resp.status_code == 200:
json = loads(resp.text)
for _ in json:
# check if user is enrolled in the course per the map_dict keys
if _['role'] in map_dict:
user_id, course_id, enroll_id = [_['user_id'], _['course_id'], _['id']]
user_list.append(course_id)
enrollments_list.append(enroll_id)
roles_list.append(map_dict[_['role']])
# build the user enrollment dictionary for those mapped roles
if user_list: user_dict = {user_id: {"courses": user_list, "enrollments": enrollments_list, "roles": roles_list}}
else: raise Exception(f"{self.__class__.__name__}.get_enrollments. Response {resp.text} returned.")
return user_dict
except:
EpsException(__file__)
def set_enrollment(self, url, user_id, role_id, enroll_id):
"""
Sets the user enrollment for the course by deleting the original enrollment, making a new one.
@param url: String
@param user_id: Int
@param role_id: Int
@param enroll_id: Int
@raise exception: EpsException
"""
try:
# now we enroll the user in the proper role
# we keep the enrollment type blank so the role id will override the base enrollment
data = {"enrollment[user_id]": user_id, "enrollment[type]": '', "enrollment[role_id]": role_id, "enrollment[enrollment_state]": "active"}
resp = self.session.post(url, data=data, headers=self.header)
if resp.status_code == 200:
# do not change the url as we want to delete the old enrollment now
resp = self.session.delete(f"{url}/{enroll_id}", data={"task": "delete"}, headers=self.header)
if resp.status_code == 200: self.rc += 1
else: raise Exception(f"{self.__class__.__name__}.set_enrollment. Response {resp.text} returned.")
else: raise Exception(f"{self.__class__.__name__}.set_enrollment. Response {resp.text} returned.")
except:
EpsException(__file__)
# end of class
x = EpsITSyncCanvasEnrollments()
print(x.rc)
This is the flow:
- Read in the configuration .ini files, one that is global (the EpsConfiguration class) and one that named the same as this class
- Assign the configuration values to class values
- Query the database for the data set of user login ids
- Get a data set of all of the roles that currently exists in our Canvas instance
- For each user, act as that user and get all of the current and future enrollments
- Using the mapping dictionary, find each enrollment that we need to change and get the role id value from the list of roles that were grabbed earlier
- For each enrollment that is applicable for the user, enroll the user in the new role for the course and set it to active and then delete the old enrollment
And there you go. You have moved all of your applicable enrollments over to the new one without having to do it manually. Setting this script up as a regular job, depending on your needs of course, will ensure that your Canvas user role assignments don't get out of control.
