canvas_api_test.py

import requests as req
import pandas as pd
import dotenv
import os
import json
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"]
    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()
    resp = json.loads(stdout)
    print(resp)


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)