Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Synchronize Canvas assignments with Ed Lesson completions.
{
"nbformat": 4,
"nbformat_minor": 0,
"metadata": {
"colab": {
"name": "canvas-lesson-sync.ipynb",
"provenance": []
},
"kernelspec": {
"name": "python3",
"display_name": "Python 3"
}
},
"cells": [
{
"cell_type": "code",
"metadata": {
"id": "rn0bFFJ29Pvj"
},
"source": [
"!pip install --upgrade -q aiohttp[speedups] nest_asyncio canvasapi\n",
"\n",
"import aiohttp\n",
"import asyncio\n",
"import io\n",
"import pandas as pd\n",
"import nest_asyncio\n",
"\n",
"nest_asyncio.apply()\n",
"\n",
"from canvasapi import Canvas\n",
"\n",
"api_key = \"\"\n",
"canvas = Canvas(\"https://canvas.uw.edu\", api_key)\n",
"token = input(\"Ed token: \")"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "code",
"metadata": {
"id": "rlVD6DFAlJaP"
},
"source": [
"# Canvas course (URL identifier)\n",
"course = canvas.get_course(...)\n",
"\n",
"# Ed lesson identifier to Canvas assignment (URL identifier)\n",
"lessons = {\n",
" ed_id: course.get_assignment(canvas_id)\n",
" for ed_id, canvas_id in [\n",
" (..., ...),\n",
" ]\n",
"}\n",
"\n",
"# Ed emails to Canvas ID\n",
"students = {\n",
" # Map Canvas emails to Ed emails\n",
" {\n",
" }.get(user.email, user.email): user.id\n",
" for user in course.get_users(enrollment_type=[\"student\"])\n",
"}\n",
"MAP_TEST_STUDENT = False\n",
"if MAP_TEST_STUDENT:\n",
" students[\"...\"] = ..."
],
"execution_count": 2,
"outputs": []
},
{
"cell_type": "code",
"metadata": {
"id": "D03IhDxb7Hkf"
},
"source": [
"# https://canvas.instructure.com/doc/api/submissions.html#method.submissions_api.bulk_update\n",
"async def completions(session, lesson):\n",
" async with session.post(\n",
" f\"https://us.edstem.org/api/lessons/{lesson}/results.csv\",\n",
" params={\n",
" \"numbers\": 0,\n",
" \"scores\": 0,\n",
" \"students\": int(not MAP_TEST_STUDENT),\n",
" \"completions\": 1,\n",
" \"strategy\": \"best\",\n",
" \"ignore_late\": 0,\n",
" \"late_no_points\": 0,\n",
" \"tutorial\": \"\",\n",
" \"tz\": \"America/Los_Angeles\",\n",
" },\n",
" data={\"_token\": token}\n",
" ) as response:\n",
" df = pd.read_csv(io.StringIO(await response.text()), index_col=\"email\").iloc[:, 3:].notna()\n",
" return {\n",
" students[email]: {\"posted_grade\": int(completed)}\n",
" for email, completed in (df.sum(axis=1) == len(df.columns)).iteritems()\n",
" if completed and (email in students or print(f\"Missing Canvas enrollment: {email}\"))\n",
" }\n",
"\n",
"async def main(lessons):\n",
" async with aiohttp.ClientSession() as session:\n",
" return await asyncio.gather(*[completions(session, lesson) for lesson in lessons])\n",
"\n",
"progress = {\n",
" lesson: assignment.submissions_bulk_update(grade_data=data)\n",
" for (lesson, assignment), data in zip(lessons.items(), asyncio.run(main(lessons)))\n",
" if data\n",
"}"
],
"execution_count": 3,
"outputs": []
}
]
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment