Bulk-update grades in Canvas based on student progress in the Ed learning management system. Requires: aiohttp[speedups] canvasapi nest_asyncio pandas requests

canvas_sync.py

import aiohttp
import asyncio
import canvasapi
import itertools
import json
import pandas as pd
import requests
import nest_asyncio

nest_asyncio.apply()

from io import StringIO
from typing import Any, NamedTuple

ED_COURSE_ID: int = ...
ED_API_TOKEN: str = ...
ED_API_HEADERS: dict[str, str] = {"Authorization": f"Bearer {ED_API_TOKEN}"}
CANVAS_COURSE_ID: int = ...
CANVAS_API_KEY: str = ...

canvas: canvasapi.Canvas = canvasapi.Canvas(..., CANVAS_API_KEY)
course: canvasapi.course.Course = canvas.get_course(CANVAS_COURSE_ID)

# Ed internal user IDs to Ed emails
users: dict[int, str] = {
    user["id"]: user["email"]
    for user in json.loads(
        requests.get(
            f"https://us.edstem.org/api/courses/{ED_COURSE_ID}/admin",
            headers=ED_API_HEADERS
        ).text
    )["users"]
}

# Ed emails to Canvas ID
students: dict[str, int] = {
    # Map Canvas emails to Ed emails
    {
    }.get(user.email, user.email): user.id
    for user in course.get_users(enrollment_type=["student"])
}
TEST_STUDENT: bool = True
if TEST_STUDENT:
    students[...] = ...


class LessonPrep(NamedTuple):
    assignment: canvasapi.assignment.Assignment
    ed_lesson_ids: tuple[int, ...]

    async def grade_data(self, session: aiohttp.ClientSession) -> dict[str, dict]:
        dfs: list[pd.DataFrame] = []
        for ed_lesson_id in self.ed_lesson_ids:
            async with session.post(
                f"https://us.edstem.org/api/lessons/{ed_lesson_id}/results.csv",
                params={
                    "numbers": 0,
                    "scores": 0,
                    "students": int(not TEST_STUDENT),
                    "completions": 1,
                    "strategy": "best",
                    "ignore_late": 0,
                    "late_no_points": 0,
                    "tutorial": "",
                    "tz": "America/Los_Angeles",
                },
            ) as response:
                df: pd.DataFrame = pd.read_csv(StringIO(await response.text())).set_index("email")
                if TEST_STUDENT:
                    df = df.filter(students, axis=0)
                start: int = df.columns.get_loc("total score") + 1
                end: int = df.columns.get_loc("⏸  Pause and 🧠 Think") + 1
                dfs.append(df.iloc[:, start:end].notna().sum(axis=1) / (end - start))
        result: pd.Series = pd.concat(dfs).groupby(level=0).sum() / len(self.ed_lesson_ids)
        result.index = result.index.map(students)
        return result.to_frame(name="posted_grade").to_dict("index")


class Discussion(NamedTuple):
    assignment: canvasapi.assignment.Assignment
    ed_discussion_id: int

    async def grade_data(self, session: aiohttp.ClientSession) -> dict[str, dict]:
        async with session.get(
            f"https://us.edstem.org/api/threads/{self.ed_discussion_id}"
        ) as response:
            data: list[dict[str, Any]] = json.loads(await response.text())["users"]
            emails: list[str] = [users[user["id"]] for user in data]
            return {
                students[email]: {"posted_grade": "pass"}
                for email in emails
                if email in students
            }


class Checkpoint(NamedTuple):
    assignment: canvasapi.assignment.Assignment
    ed_lesson_id: int

    async def grade_data(self, session: aiohttp.ClientSession) -> dict[str, dict]:
        async with session.post(
            f"https://us.edstem.org/api/lessons/{self.ed_lesson_id}/results.csv",
            params={
                "numbers": 0,
                "scores": 0,
                "students": int(not TEST_STUDENT),
                "completions": 1,
                "strategy": "best",
                "ignore_late": 0,
                "late_no_points": 0,
                "tutorial": "",
                "tz": "America/Los_Angeles",
            },
        ) as response:
            df: pd.DataFrame = pd.read_csv(StringIO(await response.text())).set_index("email")
            if TEST_STUDENT:
                df = df.filter(students, axis=0)
            start: int = df.columns.get_loc("total score") + 1
            df = df.iloc[:, start:].notna().sum(axis=1) / (len(df.columns) - start)
            df.index = df.index.map(students)
            return df.to_frame(name="posted_grade").to_dict("index")


class Assessment(NamedTuple):
    assignment: canvasapi.assignment.Assignment
    ed_challenge_id: int
    rubrics: dict[str, dict[str, float]] = {
        "Behavior": {"Exemplary": 0.55, "Satisfactory": 0.40, "Not yet": 0.25},
        "Concepts": {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
        "Quality":  {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
        "Testing":  {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
        "Writeup":  {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
    }

    def grade_from(self, email: str, **rubric_ratings: str) -> dict[str, Any]:
        result: dict = {
            "posted_grade": 0.0,
            "rubric_assessment": {},
        }
        for rubric, rating in rubric_ratings.items():
            result["posted_grade"] += self.rubrics.get(rubric, {}).get(rating, 0.0)
            try:
                criterion = next(c for c in self.assignment.rubric if c["description"] == rubric)
                crit_rating = next(r for r in criterion["ratings"] if r["description"] == rating)
            except StopIteration:
                print(f"{self.assignment.name} - {email}")
                raise
            result["rubric_assessment"][criterion["id"]] = {
                "points": crit_rating["points"],
                "rating_id": crit_rating["id"],
            }
        return result

    async def grade_data(self, session: aiohttp.ClientSession) -> dict[str, dict]:
        async with session.post(
            f"https://us.edstem.org/api/challenges/{self.ed_challenge_id}/results",
            params={
                "students": int(not TEST_STUDENT),
                "include_all": 0,
                "type": "latest-with-feedback",
                "numbers": 0,
                "scores": 0,
                "score_type": "passfail",
                "feedback": 1,
                "tz": "America/Los_Angeles",
            },
        ) as response:
            df: pd.DataFrame = pd.read_csv(StringIO(await response.text())).set_index("email")
            if TEST_STUDENT:
                df = df.filter(students, axis=0)
            return {
                students[email]: self.grade_from(email, **rubric_ratings)
                for email, rubric_ratings in df.filter(self.rubrics).to_dict("index").items()
            }


class Late:
    def __init__(self, work: Any) -> None:
        self.assignment: canvasapi.assignment.Assignment = work.assignment
        self.work: Any = work

    def only_for(self, *emails: str) -> Late:
        self.allowed_students: set[int] = set(students[email] for email in emails)
        return self

    async def grade_data(self, session: aiohttp.ClientSession) -> dict[str, dict]:
        return {
            student: grade
            for student, grade in (await self.work.grade_data(session)).items()
            if student in self.allowed_students
        }


async def main(*coursework: Any) -> dict[str, dict[str, dict]]:
    async with aiohttp.ClientSession(headers=ED_API_HEADERS) as session:
        assignments: list[int] = [work.assignment.id for work in coursework]
        tasks: list[Any] = [work.grade_data(session) for work in coursework]
        return dict(zip(assignments, await asyncio.gather(*tasks)))


gradebook: dict[str, dict[str, dict]] = asyncio.run(main(
    LessonPrep(course.get_assignment(...), (..., ..., ...)),
    Discussion(course.get_assignment(...), ...),
    Checkpoint(course.get_assignment(...), ...),
    Assessment(course.get_assignment(...), ...),
))
# https://canvas.instructure.com/doc/api/submissions.html#method.submissions_api.bulk_update
print(course.submissions_bulk_update(grade_data=gradebook).url)

Simpler, minimal example assuming a test.csv file in the following format.

Email,Behavior,Concepts,Quality,Testing
[email protected],Exemplary,Satisfactory,Not yet,Unassessable

import canvasapi
import pandas as pd

CANVAS_COURSE_ID = ...
CANVAS_API_KEY = ...
CANVAS_ASSIGNMENT_ID = ...

course = canvasapi.Canvas("https://canvas.uw.edu", CANVAS_API_KEY).get_course(CANVAS_COURSE_ID)
students = {user.email: user.id for user in course.get_users(enrollment_type=["student"])}

df = pd.read_csv("test.csv").set_index("Email")
df.index = df.index.map(students).rename("ID")

rubrics = {
    "Behavior": {"Exemplary": 0.55, "Satisfactory": 0.40, "Not yet": 0.25},
    "Concepts": {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
    "Quality":  {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
    "Testing":  {"Exemplary": 0.15, "Satisfactory": 0.15, "Not yet": 0.05},
}

def grade_from(assignment, **rubric_ratings):
    result = {
        "posted_grade": 0.0,
        "rubric_assessment": {},
    }
    for rubric, rating in rubric_ratings.items():
        result["posted_grade"] += rubrics.get(rubric, {}).get(rating, 0.0)
        criterion = next(c for c in assignment.rubric if c["description"] == rubric)
        crit_rating = next(r for r in criterion["ratings"] if r["description"] == rating)
        result["rubric_assessment"][criterion["id"]] = {
            "points": crit_rating["points"],
            "rating_id": crit_rating["id"],
        }
    return result

gradebook = {
    CANVAS_ASSIGNMENT_ID: {
        id: grade_from(course.get_assignment(CANVAS_ASSIGNMENT_ID), **rubric_ratings)
        for id, rubric_ratings in df.to_dict("index").items()
    }
}
# https://canvas.instructure.com/doc/api/submissions.html#method.submissions_api.bulk_update
print(course.submissions_bulk_update(grade_data=gradebook).url)