canvas_api_test.py

import requests as req
import pandas as pd
import dotenv
import os
import json
import re
import subprocess as sp

from time import mktime
from datetime import datetime
from collections import OrderedDict

# api documentation:
# https://bcourses.berkeley.edu/doc/api/

def upload(fp, sub_url, bcourse, bcourses_path=None):
    """
    Uploads a file on a local machine to bcourses with optional folder path
    option.
    """
    # Get file token for upload
    name = os.path.split(fp)[-1]
    filesize = os.path.getsize(fp)
    payload = {u'name':unicode(name), u'size':unicode(filesize)}
    if bcourses_path:
        payload[u'parent_folder_path'] = unicode(bcourses_path)
    resp = req.post(sub_url, data=payload, headers=bcourse.header).json()

    # Upload file data -- the -L tells curl to follow redirects
    cmd = ["curl", "{}".format(resp['upload_url']), "-L", "-i"]
    for param, val in resp['upload_params'].items():
        cmd.extend(["-F", "{}={}".format(param, val)])
    cmd.extend(["-F", "file=@{}".format(fp)])
    p = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.PIPE)
    stdout, _ = p.communicate()
    loc = stdout.split('Location: ')[1].split('\n')[0]
    print(loc)

    # Confirm upload's success
    cmd = ["curl", "-X", "POST {}".format(loc), "-H", '{}'.format('Content-Length: 0'), "-H", 'Authorization: {}'.format(bcourse.header['Authorization'])]
    p = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.PIPE)
    stdout, _ = p.communicate()
    print(stdout)

class BCourses(object):
    def __init__(self, **kwargs):
        if 'course' not in kwargs.keys():
            self.course = u'COG SCI 131'
        else:
            self.course = kwargs['course']

        dotenv.load_dotenv('.env')
        oauth = os.environ.get('TOKEN')
        self.header = {'Authorization': 'Bearer ' + oauth}
        self.params = {'per_page': 100}
        self.api = 'https://bcourses.berkeley.edu/api/v1/courses'
        self.course_id = str(self.get_course_id(self.course))


    def get_user_ids(self):
        enroll = os.path.join(self.api, self.course_id, 'enrollments')
        linx = req.head(enroll, params=self.params, headers=self.header).links
        current = linx['current']['url']
        resp = req.get(current, params=self.params, headers=self.header)

        users = pd.DataFrame(columns=['name', 'user_id', 'login_id', 'role'])
        for idx,ss in enumerate(resp.json()):
            users.loc[idx,'name'] = ss['user']['sortable_name'].upper()
            users.loc[idx,'user_id'] = ss['user']['id']
            users.loc[idx,'login_id'] = ss['user']['login_id']
            users.loc[idx,'role'] = ss['role']
        count = idx + 1

        # Deal w. pagination; responses are capped at 100 entries per page
        while 'next' in linx.keys():
            nex = linx['next']['url']
            resp = req.get(nex, params=self.params, headers=self.header)
            for idx,ss in enumerate(resp.json()):
                users.loc[count + idx, 'name'] = ss['user']['sortable_name']
                users.loc[count + idx, 'user_id'] = ss['user']['id']
                users.loc[count + idx, 'login_id'] = ss['user']['login_id']
                users.loc[count + idx, 'role'] = ss['role']
            linx = req.head(nex, params=self.params, headers=self.header).links
            count = count + idx + 1
        return users.sort_index(axis=0, by='name').reset_index(drop=True)


    def upload_comments(self, assgn_name='PS0'):
        """ UNFINISHED -- Error using upload """
        assgn_id = self.get_assignment_id(assgn_name)
        users = self.get_user_ids()

        # TODO: only loop through users we have comment files for
        for u_id in users.user_id:
            sub = os.path.join(self.api, self.course_id, 'assignments', assgn_id, 'submissions', u_id, 'comments', 'files')

            c_path = '' # path to comment file
            upload(c_path, sub, self)


    def upload_file(self, fp, bcourses_path=None):
        """ UNFINISHED -- Error using upload """
        sub = os.path.join(self.api, self.course_id, 'files')
        upload(fp, sub, self, bcourses_path)


    def get_course_id(self, course):
        resp = req.get(self.api, params=self.params, headers=self.header)
        for cc in resp.json():
            if cc['course_code'] == course:
                return cc['id']
        raise KeyError('Unable to find course %s'%course)


    def get_assignment_id(self, name='PS0'):
        assgn_url = os.path.join(self.api, self.course_id, 'assignments')
        resp = req.get(assgn_url, params=self.params, headers=self.header)
        for assgn in resp.json():
            if assgn['name'] == name:
                return assgn['id']
        raise KeyError('Unable to find assignment %s'%name)