Canvas subscription script.

_readme.md

Here's the script I wrote that automatically subscribes you to all forum threads. It's built to be run as a cron job.

You'll need to generate an API token for it which can be done at the bottom of the profile settings page. It uses canvas's REST API to do the actual subscribing, the documentation for the api is here.

I also hate Canvas's WYSIWYG editor, so I wrote some scripts to render markdown, and then inline the styles. Typically I'll write my posts in a text file, and then run:

$ fpost name_of_file.md | pbcopy    # pbcopy is an OSX thing, xclip -in works for xwindows.

and then just post the raw HTML into the WYSIWYG's 'html entry' portion.

Here's the source for the fpost command above, as well as the script is uses to inline styles inlinebody:

• https://github.com/Joshkunz/binfiles/blob/master/fpost
• https://github.com/Joshkunz/binfiles/blob/master/inlinebody

See also:

• https://pypi.python.org/pypi/pynliner
• http://johnmacfarlane.net/pandoc/

subscribe.py

from requests.auth import AuthBase
from requests import Session
import os.path
import sys

CANVAS_API = "https://utah.instructure.com/api/v1/"
TOKEN_PATH = "$HOME/.canvas_token"

# Expand the abbreviated path
TOKEN_PATH = os.path.expanduser(TOKEN_PATH)
TOKEN_PATH = os.path.expandvars(TOKEN_PATH)
TOKEN_PATH = os.path.abspath(TOKEN_PATH)

class PrefixSession(Session):

    def __init__(self, url_prefix):
        self.prefix = url_prefix
        super(PrefixSession, self).__init__()

    @classmethod
    def path_prepend(cls, base, extension):
        "prepend 'base' onto the path 'extension'"
        if base[-1] == "/": base = base[:-1]
        if extension[0] == "/": extension = extension[1:]
        return "/".join((base, extension))

    def request(self, method, url, *args, **kwargs):
        "Prefix the url with the supplied prefix when a request is made"
        url = self.path_prepend(self.prefix, url)
        return super(PrefixSession, self).request(method, url, *args, **kwargs)


class CanvasAuth(AuthBase):

    def __init__(self, access_token, base_url=None):
        self.token = access_token

    def __call__(self, request):
        request.headers["Authorization"] = " ".join(("Bearer", self.token))
        return request

# Helper methods 
def load_token(token_file):
    with open(token_file, 'r') as tf:
        token = tf.readline().strip()
        return token if token else None

def canvas_courses(canvas):
    return canvas.get("/courses", params={"include[]": "term"})

def canvas_topics(canvas, course_id):
    endpoint = "/courses/{course_id}/discussion_topics"
    return canvas.get(endpoint.format(course_id=course_id))

def canvas_topic_subscribe(canvas, course_id, topic_id):
    endpoint = "/courses/{course_id}/discussion_topics/{topic_id}/subscribed"
    return canvas.put(endpoint.format(course_id=course_id, topic_id=topic_id))

# ensure the file exists
if not os.path.exists(TOKEN_PATH):
    sys.stderr.write("No token file found at the supplied path.\n")
    sys.exit(-1)

canvas = PrefixSession(CANVAS_API)
canvas.auth = CanvasAuth(load_token(TOKEN_PATH))

courses = canvas_courses(canvas) 
for course in courses.json():
    # If the course has been completed, skip it
    if course["term"]["end_at"] != None:
        continue

    # Track wether or not we've printed the course's name
    printed_course = False

    topics = canvas_topics(canvas, course["id"]) 
    for topic in topics.json():
        if topic["subscribed"] == True: continue

        # print the course if we're subscribing to a topic
        #if not printed_course: 
        #   print course["name"]
        #   printed_course = True

        canvas_topic_subscribe(canvas, course["id"], topic["id"])
        #print "\tSubscribed: {0}".format(topic["title"])