-
Notifications
You must be signed in to change notification settings - Fork 2
/
canvas_tools.py
153 lines (137 loc) · 5.56 KB
/
canvas_tools.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import json
import simplejson
import requests
import yaml
import os
import sys
import time
from datetime import datetime
try:
from tqdm import tqdm
except:
tqdm = None
def yaml_load(path):
with open(path) as settings_file:
return yaml.load(settings_file)
settings = {
'courses': {},
'canvas-token': '',
'defaults': {},
'canvas-url': 'https://vt.instructure.com/api/v1/'
}
courses = {}
defaults = {}
def load_settings(path='settings.yaml', create_if_not_exists=True):
global courses, defaults
# Create settings file if it doesn't exist
if not os.path.exists(path):
if create_if_not_exists:
with open(path, 'w') as settings_file:
yaml.dump(settings, settings_file)
print("A settings.yaml file was created. Please add your token and courses.")
sys.exit()
else:
raise Exception("The settings file was not found: "+repr(path))
# Load in the settings file
new_settings = yaml_load(path)
settings.update(new_settings)
# Shortcut to access courses
courses = settings['courses']
defaults = settings['defaults']
def get_courses():
return settings['courses']
def get_setting(setting, course=None):
if course is None:
return defaults[setting]
if course in courses:
if setting in courses[course]:
return courses[course][setting]
return defaults[setting]
raise Exception("Course not found in settings.yaml: {course}".format(course=course))
def _canvas_request(verb, command, course, data, all, params, result=list,
estimated=None):
try:
if data is None:
data = {}
if params is None:
params = {}
if course == 'default':
course = get_setting('course')
next_url = get_setting('canvas-url', course=course)
if course != None:
course_id = courses[course]['id']
next_url += 'courses/{course_id}/'.format(course_id=course_id)
next_url += command
data['access_token'] = get_setting('canvas-token')
if all:
data['per_page'] = 100
final_result = []
if tqdm is not None and estimated is not None:
pbar = tqdm(total=estimated/100)
while True:
if tqdm is not None and estimated is not None:
pbar.update(1)
response = verb(next_url, data=data, params=params)
if 'error' in response.json():
raise(Exception(response.text))
if result == list:
final_result += response.json()
elif result == dict:
final_result.append(response.json())
else:
final_result = response
if 'next' in response.links:
next_url = response.links['next']['url']
else:
if tqdm is not None and estimated is not None:
pbar.close()
return final_result
else:
response = verb(next_url, data=data, params=params)
if result == list:
return response.json()
elif result == dict:
return [response.json()]
except simplejson.decoder.JSONDecodeError:
raise Exception("{}\n{}".format(response, next_url))
def get(command, course='default', data=None, all=False, params=None, result=list, estimated=None):
return _canvas_request(requests.get, command, course, data, all, params,
result=result, estimated=estimated)
def post(command, course='default', data=None, all=False, params=None, result=list, estimated=None):
return _canvas_request(requests.post, command, course, data, all, params,
result=result, estimated=estimated)
def put(command, course='default', data=None, all=False, params=None, result=list, estimated=None):
return _canvas_request(requests.put, command, course, data, all, params,
result=result, estimated=estimated)
def delete(command, course='default', data=None, all=False, params=None, result=list, estimated=None):
return _canvas_request(requests.delete, command, course, data, all, params,
result=result, estimated=estimated)
def progress_loop(progress_id, DELAY=3):
attempt = 0
while True:
result = _canvas_request(requests.get, 'progress/{}'.format(progress_id),
None, {'_dummy_counter': attempt},
False, None, dict)[0]
if result['workflow_state'] == 'completed':
return True
elif result['workflow_state'] == 'failed':
return False
else:
print("In progress:", result['workflow_state'], result['message'],
str(int(round(result['completion']*10))/10)+"%")
if not hasattr(result, 'from_cache') or not result.from_cache:
time.sleep(DELAY)
attempt += 1
def download_file(url, destination):
data = {'access_token': get_setting('canvas-token')}
r = requests.get(url)
f = open(destination, 'wb')
for chunk in r.iter_content(chunk_size=512 * 1024):
if chunk: # filter out keep-alive new chunks
f.write(chunk)
f.close()
CANVAS_DATE_STRING = "%Y-%m-%dT%H:%M:%SZ"
def from_canvas_date(d1):
return datetime.strptime(d1, CANVAS_DATE_STRING)
def to_canvas_date(d1):
return d1.strftime(CANVAS_DATE_STRING)