Update student email notifications as an admin in Canvas LMS

update_student_notifications.py

import csv
import re
import requests
import concurrent.futures
import time
# pip install tqdm for progress monitoring
from tqdm import tqdm
from functools import partial

"""
There is no way for a Canvas admin to update a user's preferences from
the admin console. This script will override a student's notification
preferences by masquerading a PUT request through the Canvas API.

Additionally, there is no way to programmatically get current enrollments
from an account or subaccount level. This script relies on CSV created from the
subaccount admin console.

Subaccount > Settings > Reports > Provisioning
Set params to the current grading term for the subaccount and Enrollments.

The script filters out teacher enrollments and only updates students in memory. 

BEFORE YOU BEGIN:
- Change your CSV location on line 43.
- Set your Canvas URL on lines 73 and 99

See function definitions for params and valid options.
"""

headers = {
    'Authorization': 'Bearer your_auth_token'
}

def process_student_id(student):
    # Get their communication channel prefs
    pref_id = get_prefs(student)

    try:
        update = update_prefs(student, pref_id)
        return update
    except Exception as e:
        print(e)

def main():
    """
    Update Canvas user notification preferences as an admin.
    """
    unique = set()
    data = []
    with open('your.csv', 'r') as inp:
        for row in csv.reader(inp):
            if re.search("student", row[4]):
                unique.add(int(row[2]))
    
    # Rate limiting kicked in at 4 concurrent workers. Do some testing before you start.
    # At 3 workers, this processed 2 students/sec to update three preferences.
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        with tqdm(total=len(unique)) as progress:
            futures = []
            for student in unique:
                future = executor.submit(process_student_id, student)
                future.add_done_callback(lambda p: progress.update())
                futures.append(future)
            
            results = [future.result() for future in futures]

def get_prefs(student_id):
    """
    Notification preferences are sorted into types like "email", "push", etc.
    This function returns the ID for the student email preference.
    :param: Int student Canvas ID
    :returns: Int communication_channel ID
    """
    url = f"https://yourURL.instructure.com/api/v1/users/{student_id}/communication_channels"
    resp = requests.request("GET", url, headers=headers)

    # build a list of channel IDs
    for channel in resp.json():
        # find the ID of the email pref
        if channel['type'] == 'email':
            return channel['id']

def update_prefs(student_id, channel_id):
    """
    Defines a list of communication types to update.
    :param: Int student_id
    :param: Int channel_id
    :returns: List <requests.Response>
    
    List your preferences to see valid string types.
    https://canvas.instructure.com/doc/api/notification_preferences.html#method.notification_preferences.index
    """
    types = ["new_announcement", "submission_comment", "conversation_message"]
    frequency = "" # 'immediately', 'daily', 'weekly', 'never'
    responses = []

    for msg_type in types:
        # Update a notification
        # https://canvas.instructure.com/doc/api/notification_preferences.html#method.notification_preferences.update
        url = f"https://yourURL.instructure.com/api/v1/users/self/communication_channels/{channel_id}/notification_preferences/{msg_type}?as_user_id={student_id}&notification_preferences[frequency]={frequency}"
        resp = requests.request("PUT", url, headers=headers)

        responses.append(resp)
    
    return responses

if __name__ == "__main__":
    start = time.perf_counter()
    print("Starting...")
    main()
    print("Finished")
    finish = time.perf_counter()
    print(f'Finished in {round(finish-start, 2)} second(s)')