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
154
155
156
157
158
159
160
161
162
163
164
|
#!/usr/bin/env python3
import requests
import logging
import dataclasses
from typing import List
log = logging.getLogger("muddle.moodle")
def get_token(url, user, password):
token_url = f"{url}/login/token.php"
data = {
"username": user,
"password": password,
"service": "moodle_mobile_app"
}
log.debug(f"requesting token with POST to {api_url} with DATA {data}")
return requests.post(token_url, data=data)
class RestApi:
"""
Magic REST API wrapper (ab)using lambdas
"""
def __init__(self, instance_url, token=None):
self._url = instance_url
if token:
self._token = token
def __getattr__(self, key):
return lambda **kwargs: RestApi._call(self._url, self._token, str(key), **kwargs)
def _call(self, function, **kwargs):
return RestApi._call(self._url, self._token, kwargs)
@staticmethod
def _call(url, token, function, **kwargs):
api_url = f"{url}/webservice/rest/server.php?moodlewsrestformat=json"
data = {"wstoken": token, "wsfunction": function}
for k, v in kwargs.items():
data[str(k)] = v
log.debug(f"calling api with POST to {api_url} with DATA {data}")
try:
req = requests.post(api_url, data=data)
req.raise_for_status()
except requests.HTTPError:
log.warn("Error code returned by HTTP(s) request")
except (requests.ConnectionError, requests.Timeout, requests.ReadTimeout) as e:
log.error(f"Failed to connect for POST request:\n{str(e)}")
finally:
return req
class MoodleInstance:
"""
A more frendly API that wraps around the raw RestApi
"""
def __init__(self, url, token):
self.api = RestApi(url, token)
def get_userid(self):
req = self.api.core_webservice_get_site_info()
return req.json()["userid"]
def get_enrolled_courses(self):
req = self.api.core_enrol_get_users_courses(userid=self.get_userid())
for c in req.json():
yield Course._fromdict(c)
# A bare minimum impl of Moodle SCHEMA
# Beware that lots of parameters have been omitted
class SchemaObj:
@classmethod
def _fromdict(cls, d):
"""
Creates a schema object from a dictionary, if the dictionary contains
keys that are not present in the schema object they will be ignored
"""
if cls is SchemaObj:
raise TypeError("Must be used in a subclass")
fields = [f.name for f in dataclasses.fields(cls)]
filtered = {k: v for k, v in d.items() if k in fields}
return cls(**filtered)
@dataclasses.dataclass
class Course(SchemaObj):
"""
A course, pretty self explanatory
https://www.examulator.com/er/output/tables/course.html
"""
id: int
shortname: str
fullname: str
summary: str
startdate: int
enddate: int
def get_sections(self, api):
req = api.core_course_get_contents(courseid=self.id)
for sec in req.json():
# rest api response does not contain course id
sec["course"] = self.id
yield Section._fromdict(sec)
@dataclasses.dataclass
class Section(SchemaObj):
"""
Sections of a course
https://www.examulator.com/er/output/tables/course_sections.html
"""
id: int
course: int
section: int
name: str
summary: str
visible: bool
modules: List
@dataclasses.dataclass
class Module(SchemaObj):
"""
Modules of a Course, they are grouped in sections
https://www.examulator.com/er/output/tables/course_modules.html
"""
id: int
course: int
module: int
section: int
@dataclasses.dataclass
class Folder(SchemaObj):
"""
Resource type that holds files
"""
class ApiHelper:
def __init__(self, api):
self.api = api
def get_userid(self):
req = self.api.core_webservice_get_site_info()
if req:
return req.json()["userid"]
else:
return None
def get_file(self, url, local_path):
with requests.post(url, data={"token": self.api._token}, stream=True) as r:
r.raise_for_status()
with open(local_path, "wb") as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
|
