Synchronize Canvas assignments with Ed Lesson completions.

canvas_lesson_sync.ipynb

{
  "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": []
    }
  ]
}

No longer maintained. Use canvas_sync.py instead!