aboutsummaryrefslogtreecommitdiffstats
path: root/muddle/moodle.py
blob: 870de8a0ef0174bcceb421b117bf23474c8c2cf2 (plain)
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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
#!/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 {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)
        self.userid = None

    def get_userid(self):
        if self.userid is None:
            req = self.api.core_webservice_get_site_info()
            self.userid = req.json()["userid"]

        return self.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)


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)


# A bare minimum impl of Moodle SCHEMA
# This is an experiment and not currently in use!
# 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 s in req.json():
            # rest api response does not contain course id
            s["course"] = self.id
            yield Section._fromdict(s)


@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

    def get_modules(self):
        for m in self.modules:
            yield Module._fromdict(m)


@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
    name: str


@dataclasses.dataclass
class File(SchemaObj):
    filename: str
    fileurl: str


@dataclasses.dataclass
class Folder(SchemaObj):
    pass


@dataclasses.dataclass
class ExternalLink(SchemaObj):
    pass