Skip to content

Instantly share code, notes, and snippets.

@ddbourgin
Last active September 24, 2015 03:33
Show Gist options
  • Save ddbourgin/4dff2fa5e4a10fe6c798 to your computer and use it in GitHub Desktop.
Save ddbourgin/4dff2fa5e4a10fe6c798 to your computer and use it in GitHub Desktop.
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment