From 6e079f88e556e2cd142ec372bafb26435e6c9fd1 Mon Sep 17 00:00:00 2001
From: Nao Pross <np@0hm.ch>
Date: Fri, 12 Feb 2021 00:11:43 +0100
Subject: Partially refractor muddle.moodle and add test
---
muddle/moodle.py | 150 +++++++++++++++++++++++++++++++++++++++++++++----------
test/moodle.py | 25 ++++++++++
2 files changed, 148 insertions(+), 27 deletions(-)
create mode 100644 test/moodle.py
diff --git a/muddle/moodle.py b/muddle/moodle.py
index 7f3ccf5..a9209d7 100644
--- a/muddle/moodle.py
+++ b/muddle/moodle.py
@@ -1,15 +1,14 @@
#!/usr/bin/env python3
import requests
import logging
+import dataclasses
-log = logging.getLogger("muddle.moodle")
+from typing import List
-#
-# magic moodle api wrapper
-#
+log = logging.getLogger("muddle.moodle")
-def request_token(url, user, password):
+def get_token(url, user, password):
token_url = f"{url}/login/token.php"
data = {
"username": user,
@@ -20,32 +19,129 @@ def request_token(url, user, password):
return requests.post(token_url, data=data)
-def api_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()
- return req
- except requests.HTTPError:
- log.warn(f"Error code returned by HTTP(s) request")
- return req
- except (requests.ConnectionError, requests.Timeout, requests.ReadTimeout) as e:
- log.error(f"Failed to connect for POST request:\n{str(e)}")
- return None
-
-
class RestApi:
- def __init__(self, instance_url, token):
+ """
+ Magic REST API wrapper (ab)using lambdas
+ """
+ def __init__(self, instance_url, token=None):
self._url = instance_url
- self._token = token
+ if token:
+ self._token = token
def __getattr__(self, key):
- return lambda **kwargs: api_call(self._url, self._token, str(key), **kwargs)
+ 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:
diff --git a/test/moodle.py b/test/moodle.py
new file mode 100644
index 0000000..3669ef8
--- /dev/null
+++ b/test/moodle.py
@@ -0,0 +1,25 @@
+import pytest
+
+import pathlib
+import configparser
+
+from muddle import paths
+from muddle import moodle
+
+config_file = pathlib.Path(paths.default_config_file)
+if not config_file.is_file():
+ log.error(f"cannot read {config_file}")
+ sys.exit(1)
+
+config = configparser.ConfigParser()
+config.read(config_file)
+
+
+class TestMoodleInstance:
+ server = moodle.MoodleInstance(config["server"]["url"], config["server"]["token"])
+
+ def test_get_userid(self):
+ assert self.server.get_userid() != None
+
+ def test_get_enrolled_courses(self):
+ assert type(next(self.server.get_enrolled_courses())) == moodle.Course
--
cgit v1.2.1