Skip to content

Moodle API Client

REST API client for Moodle Web Services.

mograder.transport.moodle_api

Moodle REST Web Services API client for mograder.

MoodleAPIError

Bases: Exception

Raised when a Moodle API call returns an error.

Source code in src/mograder/transport/moodle_api.py 
14
15
16
17
18
19
class MoodleAPIError(Exception):
    """Raised when a Moodle API call returns an error."""

    def __init__(self, message: str, error_code: str | None = None):
        self.error_code = error_code
        super().__init__(message)

MoodleAPIClient

Client for Moodle's REST Web Services API.

Source code in src/mograder/transport/moodle_api.py 
 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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
class MoodleAPIClient:
    """Client for Moodle's REST Web Services API."""

    def __init__(self, url: str, token: str):
        self.base_url = url.rstrip("/")
        self.endpoint = f"{self.base_url}/webservice/rest/server.php"
        self.upload_endpoint = f"{self.base_url}/webservice/upload.php"
        self.token = token

    def _call(self, wsfunction: str, **params) -> dict | list:
        """Make a Moodle web service API call."""
        data = {
            "wstoken": self.token,
            "moodlewsrestformat": "json",
            "wsfunction": wsfunction,
            **params,
        }
        resp = requests.post(self.endpoint, data=data, timeout=30)
        resp.raise_for_status()
        result = resp.json()
        if isinstance(result, dict) and "exception" in result:
            raise MoodleAPIError(
                result.get("message", result.get("exception", "Unknown error")),
                error_code=result.get("errorcode"),
            )
        return result

    def get_site_info(self) -> dict:
        """Get site info for the authenticated user.

        Returns dict with: userid, username, fullname, sitename.
        """
        result = self._call("core_webservice_get_site_info")
        return {
            "userid": result["userid"],
            "username": result["username"],
            "fullname": result["fullname"],
            "sitename": result["sitename"],
        }

    def get_assignments(self, course_id: int) -> list[dict]:
        """Get assignments for a course.

        Returns a flat list of assignment dicts with keys:
        id, cmid, name, duedate, intro, introattachments.
        """
        result = self._call("mod_assign_get_assignments", **{"courseids[0]": course_id})
        assignments = []
        for course in result.get("courses", []):
            for assign in course.get("assignments", []):
                assignments.append(
                    {
                        "id": assign["id"],
                        "cmid": assign.get("cmid"),
                        "name": assign["name"],
                        "duedate": assign.get("duedate", 0),
                        "intro": assign.get("intro", ""),
                        "introattachments": assign.get("introattachments", []),
                    }
                )
        return assignments

    def download_file(self, file_url: str, dest: Path) -> Path:
        """Download a file from Moodle, appending the token for auth.

        Automatically converts ``pluginfile.php`` URLs to
        ``webservice/pluginfile.php`` so token authentication works.
        """
        # Token auth requires the webservice endpoint
        if (
            "/pluginfile.php/" in file_url
            and "/webservice/pluginfile.php/" not in file_url
        ):
            file_url = file_url.replace(
                "/pluginfile.php/", "/webservice/pluginfile.php/"
            )
        sep = "&" if "?" in file_url else "?"
        authed_url = f"{file_url}{sep}token={self.token}"
        resp = requests.get(authed_url, stream=True, timeout=60)
        resp.raise_for_status()
        dest.parent.mkdir(parents=True, exist_ok=True)
        with open(dest, "wb") as f:
            for chunk in resp.iter_content(chunk_size=8192):
                f.write(chunk)
        return dest

    def upload_file(self, filepath: Path, itemid: int = 0) -> int:
        """Upload a file to the user's draft area.

        When *itemid* is 0 (default), a new draft area is created.
        Pass a non-zero *itemid* to append to an existing draft area.

        Returns the draft area itemid.
        """
        with open(filepath, "rb") as f:
            resp = requests.post(
                self.upload_endpoint,
                params={"token": self.token},
                files={"file_1": (filepath.name, f)},
                data={"itemid": itemid},
                timeout=60,
            )
        resp.raise_for_status()
        result = resp.json()
        if isinstance(result, dict) and "exception" in result:
            raise MoodleAPIError(
                result.get("message", "Upload failed"),
                error_code=result.get("errorcode"),
            )
        return result[0]["itemid"]

    def upload_files_to_draft(self, filepaths: list[Path]) -> int:
        """Upload multiple files to a single draft area.

        Returns the draft area itemid.
        """
        if not filepaths:
            raise ValueError("No files to upload")
        itemid = 0
        for fp in filepaths:
            itemid = self.upload_file(fp, itemid=itemid)
        return itemid

    def update_introattachments(self, cmid: int, draft_itemid: int) -> None:
        """Attach a draft area to an assignment's introattachments.

        Uses ``core_course_edit_module`` to update the assignment module.
        Raises ``MoodleAPIError`` if the API call fails (e.g. insufficient
        permissions).
        """
        self._call(
            "core_course_edit_module",
            action="update",
            id=cmid,
            **{"introattachments": draft_itemid},
        )

    def update_intro(self, cmid: int, intro_html: str) -> None:
        """Update the assignment description (intro) field.

        Uses ``core_course_edit_module`` to replace the intro HTML.
        """
        self._call(
            "core_course_edit_module",
            action="update",
            id=cmid,
            intro=intro_html,
            introformat=1,
        )

    def save_submission(self, assignment_id: int, item_id: int) -> None:
        """Save a file submission draft for an assignment."""
        self._call(
            "mod_assign_save_submission",
            assignmentid=assignment_id,
            **{
                "plugindata[files_filemanager]": item_id,
            },
        )

    def submit_for_grading(self, assignment_id: int) -> None:
        """Finalize a submission so it's visible to graders."""
        self._call(
            "mod_assign_submit_for_grading",
            assignmentid=assignment_id,
            acceptsubmissionstatement=1,
        )

    def get_submission_status(self, assignment_id: int) -> dict:
        """Get the current user's submission status, grade, and feedback.

        Returns dict with keys: status, graded, grade, feedback.
        """
        result = self._call("mod_assign_get_submission_status", assignid=assignment_id)
        sub_status = "new"
        if "lastattempt" in result:
            sub = result["lastattempt"].get("submission", {})
            sub_status = sub.get("status", "new")

        graded = False
        grade = None
        feedback_text = ""
        if "feedback" in result:
            fb = result["feedback"]
            grade_info = fb.get("grade")
            if grade_info is not None:
                graded = True
                grade = str(grade_info.get("grade", ""))
            for plugin in fb.get("plugins", []):
                if plugin.get("type") == "comments":
                    for field in plugin.get("editorfields", []):
                        if field.get("text"):
                            feedback_text = field["text"]
                            break
                    if not feedback_text:
                        for filearea in plugin.get("fileareas", []):
                            if filearea.get("text"):
                                feedback_text = filearea["text"]
                                break

        return {
            "status": sub_status,
            "graded": graded,
            "grade": grade,
            "feedback": feedback_text,
        }

    def get_submissions(self, assignment_id: int) -> list[dict]:
        """Get all submissions for an assignment.

        Returns list of dicts with keys: userid, status, files.
        Each file has: filename, fileurl, filesize.
        """
        result = self._call(
            "mod_assign_get_submissions", **{"assignmentids[0]": assignment_id}
        )
        submissions = []
        for assign in result.get("assignments", []):
            for sub in assign.get("submissions", []):
                files = []
                for plugin in sub.get("plugins", []):
                    if plugin.get("type") == "file":
                        for filearea in plugin.get("fileareas", []):
                            for f in filearea.get("files", []):
                                files.append(
                                    {
                                        "filename": f["filename"],
                                        "fileurl": f["fileurl"],
                                        "filesize": f.get("filesize", 0),
                                        "timemodified": f.get("timemodified", 0),
                                    }
                                )
                submissions.append(
                    {
                        "userid": sub["userid"],
                        "status": sub.get("status", ""),
                        "files": files,
                    }
                )
        return submissions

    def list_participants(self, assignment_id: int) -> list[dict]:
        """List participants for an assignment.

        Returns list of {id, username, fullname}.
        """
        result = self._call(
            "mod_assign_list_participants",
            assignid=assignment_id,
            groupid=0,
            filter="",
        )
        return [
            {
                "id": p["id"],
                "username": p.get("username", ""),
                "fullname": p.get("fullname", ""),
            }
            for p in result
        ]

    def save_grades(
        self,
        assignment_id: int,
        grades: list[dict],
        workflow_state: str = "",
    ) -> None:
        """Save grades for multiple students.

        Each grade dict should have: userid, grade, feedback (text).
        Uses the singular save_grade API for broader token compatibility.
        """
        for g in grades:
            params: dict = {
                "assignmentid": assignment_id,
                "userid": g["userid"],
                "grade": g["grade"],
                "attemptnumber": g.get("attemptnumber", -1),
                "addattempt": 0,
                "workflowstate": workflow_state,
                "applytoall": 0,
                "plugindata[assignfeedbackcomments_editor][text]": g.get(
                    "feedback", ""
                ),
                "plugindata[assignfeedbackcomments_editor][format]": 1,
            }
            feedback_file = g.get("feedback_file")
            if feedback_file:
                draft_itemid = self.upload_file(Path(feedback_file))
                params["plugindata[files_filemanager]"] = draft_itemid
            self._call("mod_assign_save_grade", **params)

get_site_info()

Get site info for the authenticated user.

Returns dict with: userid, username, fullname, sitename.

Source code in src/mograder/transport/moodle_api.py 
49
50
51
52
53
54
55
56
57
58
59
60
def get_site_info(self) -> dict:
    """Get site info for the authenticated user.

    Returns dict with: userid, username, fullname, sitename.
    """
    result = self._call("core_webservice_get_site_info")
    return {
        "userid": result["userid"],
        "username": result["username"],
        "fullname": result["fullname"],
        "sitename": result["sitename"],
    }

get_assignments(course_id)

Get assignments for a course.

Returns a flat list of assignment dicts with keys: id, cmid, name, duedate, intro, introattachments.

Source code in src/mograder/transport/moodle_api.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_assignments(self, course_id: int) -> list[dict]:
    """Get assignments for a course.

    Returns a flat list of assignment dicts with keys:
    id, cmid, name, duedate, intro, introattachments.
    """
    result = self._call("mod_assign_get_assignments", **{"courseids[0]": course_id})
    assignments = []
    for course in result.get("courses", []):
        for assign in course.get("assignments", []):
            assignments.append(
                {
                    "id": assign["id"],
                    "cmid": assign.get("cmid"),
                    "name": assign["name"],
                    "duedate": assign.get("duedate", 0),
                    "intro": assign.get("intro", ""),
                    "introattachments": assign.get("introattachments", []),
                }
            )
    return assignments

download_file(file_url, dest)

Download a file from Moodle, appending the token for auth.

Automatically converts pluginfile.php URLs to webservice/pluginfile.php so token authentication works.

Source code in src/mograder/transport/moodle_api.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def download_file(self, file_url: str, dest: Path) -> Path:
    """Download a file from Moodle, appending the token for auth.

    Automatically converts ``pluginfile.php`` URLs to
    ``webservice/pluginfile.php`` so token authentication works.
    """
    # Token auth requires the webservice endpoint
    if (
        "/pluginfile.php/" in file_url
        and "/webservice/pluginfile.php/" not in file_url
    ):
        file_url = file_url.replace(
            "/pluginfile.php/", "/webservice/pluginfile.php/"
        )
    sep = "&" if "?" in file_url else "?"
    authed_url = f"{file_url}{sep}token={self.token}"
    resp = requests.get(authed_url, stream=True, timeout=60)
    resp.raise_for_status()
    dest.parent.mkdir(parents=True, exist_ok=True)
    with open(dest, "wb") as f:
        for chunk in resp.iter_content(chunk_size=8192):
            f.write(chunk)
    return dest

upload_file(filepath, itemid=0)

Upload a file to the user's draft area.

When itemid is 0 (default), a new draft area is created. Pass a non-zero itemid to append to an existing draft area.

Returns the draft area itemid.

Source code in src/mograder/transport/moodle_api.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def upload_file(self, filepath: Path, itemid: int = 0) -> int:
    """Upload a file to the user's draft area.

    When *itemid* is 0 (default), a new draft area is created.
    Pass a non-zero *itemid* to append to an existing draft area.

    Returns the draft area itemid.
    """
    with open(filepath, "rb") as f:
        resp = requests.post(
            self.upload_endpoint,
            params={"token": self.token},
            files={"file_1": (filepath.name, f)},
            data={"itemid": itemid},
            timeout=60,
        )
    resp.raise_for_status()
    result = resp.json()
    if isinstance(result, dict) and "exception" in result:
        raise MoodleAPIError(
            result.get("message", "Upload failed"),
            error_code=result.get("errorcode"),
        )
    return result[0]["itemid"]

upload_files_to_draft(filepaths)

Upload multiple files to a single draft area.

Returns the draft area itemid.

Source code in src/mograder/transport/moodle_api.py 
133
134
135
136
137
138
139
140
141
142
143
def upload_files_to_draft(self, filepaths: list[Path]) -> int:
    """Upload multiple files to a single draft area.

    Returns the draft area itemid.
    """
    if not filepaths:
        raise ValueError("No files to upload")
    itemid = 0
    for fp in filepaths:
        itemid = self.upload_file(fp, itemid=itemid)
    return itemid

update_introattachments(cmid, draft_itemid)

Attach a draft area to an assignment's introattachments.

Uses core_course_edit_module to update the assignment module. Raises MoodleAPIError if the API call fails (e.g. insufficient permissions).

Source code in src/mograder/transport/moodle_api.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
def update_introattachments(self, cmid: int, draft_itemid: int) -> None:
    """Attach a draft area to an assignment's introattachments.

    Uses ``core_course_edit_module`` to update the assignment module.
    Raises ``MoodleAPIError`` if the API call fails (e.g. insufficient
    permissions).
    """
    self._call(
        "core_course_edit_module",
        action="update",
        id=cmid,
        **{"introattachments": draft_itemid},
    )

update_intro(cmid, intro_html)

Update the assignment description (intro) field.

Uses core_course_edit_module to replace the intro HTML.

Source code in src/mograder/transport/moodle_api.py 
159
160
161
162
163
164
165
166
167
168
169
170
def update_intro(self, cmid: int, intro_html: str) -> None:
    """Update the assignment description (intro) field.

    Uses ``core_course_edit_module`` to replace the intro HTML.
    """
    self._call(
        "core_course_edit_module",
        action="update",
        id=cmid,
        intro=intro_html,
        introformat=1,
    )

save_submission(assignment_id, item_id)

Save a file submission draft for an assignment.

Source code in src/mograder/transport/moodle_api.py 
172
173
174
175
176
177
178
179
180
def save_submission(self, assignment_id: int, item_id: int) -> None:
    """Save a file submission draft for an assignment."""
    self._call(
        "mod_assign_save_submission",
        assignmentid=assignment_id,
        **{
            "plugindata[files_filemanager]": item_id,
        },
    )

submit_for_grading(assignment_id)

Finalize a submission so it's visible to graders.

Source code in src/mograder/transport/moodle_api.py 
182
183
184
185
186
187
188
def submit_for_grading(self, assignment_id: int) -> None:
    """Finalize a submission so it's visible to graders."""
    self._call(
        "mod_assign_submit_for_grading",
        assignmentid=assignment_id,
        acceptsubmissionstatement=1,
    )

get_submission_status(assignment_id)

Get the current user's submission status, grade, and feedback.

Returns dict with keys: status, graded, grade, feedback.

Source code in src/mograder/transport/moodle_api.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def get_submission_status(self, assignment_id: int) -> dict:
    """Get the current user's submission status, grade, and feedback.

    Returns dict with keys: status, graded, grade, feedback.
    """
    result = self._call("mod_assign_get_submission_status", assignid=assignment_id)
    sub_status = "new"
    if "lastattempt" in result:
        sub = result["lastattempt"].get("submission", {})
        sub_status = sub.get("status", "new")

    graded = False
    grade = None
    feedback_text = ""
    if "feedback" in result:
        fb = result["feedback"]
        grade_info = fb.get("grade")
        if grade_info is not None:
            graded = True
            grade = str(grade_info.get("grade", ""))
        for plugin in fb.get("plugins", []):
            if plugin.get("type") == "comments":
                for field in plugin.get("editorfields", []):
                    if field.get("text"):
                        feedback_text = field["text"]
                        break
                if not feedback_text:
                    for filearea in plugin.get("fileareas", []):
                        if filearea.get("text"):
                            feedback_text = filearea["text"]
                            break

    return {
        "status": sub_status,
        "graded": graded,
        "grade": grade,
        "feedback": feedback_text,
    }

get_submissions(assignment_id)

Get all submissions for an assignment.

Returns list of dicts with keys: userid, status, files. Each file has: filename, fileurl, filesize.

Source code in src/mograder/transport/moodle_api.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def get_submissions(self, assignment_id: int) -> list[dict]:
    """Get all submissions for an assignment.

    Returns list of dicts with keys: userid, status, files.
    Each file has: filename, fileurl, filesize.
    """
    result = self._call(
        "mod_assign_get_submissions", **{"assignmentids[0]": assignment_id}
    )
    submissions = []
    for assign in result.get("assignments", []):
        for sub in assign.get("submissions", []):
            files = []
            for plugin in sub.get("plugins", []):
                if plugin.get("type") == "file":
                    for filearea in plugin.get("fileareas", []):
                        for f in filearea.get("files", []):
                            files.append(
                                {
                                    "filename": f["filename"],
                                    "fileurl": f["fileurl"],
                                    "filesize": f.get("filesize", 0),
                                    "timemodified": f.get("timemodified", 0),
                                }
                            )
            submissions.append(
                {
                    "userid": sub["userid"],
                    "status": sub.get("status", ""),
                    "files": files,
                }
            )
    return submissions

list_participants(assignment_id)

List participants for an assignment.

Returns list of {id, username, fullname}.

Source code in src/mograder/transport/moodle_api.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def list_participants(self, assignment_id: int) -> list[dict]:
    """List participants for an assignment.

    Returns list of {id, username, fullname}.
    """
    result = self._call(
        "mod_assign_list_participants",
        assignid=assignment_id,
        groupid=0,
        filter="",
    )
    return [
        {
            "id": p["id"],
            "username": p.get("username", ""),
            "fullname": p.get("fullname", ""),
        }
        for p in result
    ]

save_grades(assignment_id, grades, workflow_state='')

Save grades for multiple students.

Each grade dict should have: userid, grade, feedback (text). Uses the singular save_grade API for broader token compatibility.

Source code in src/mograder/transport/moodle_api.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def save_grades(
    self,
    assignment_id: int,
    grades: list[dict],
    workflow_state: str = "",
) -> None:
    """Save grades for multiple students.

    Each grade dict should have: userid, grade, feedback (text).
    Uses the singular save_grade API for broader token compatibility.
    """
    for g in grades:
        params: dict = {
            "assignmentid": assignment_id,
            "userid": g["userid"],
            "grade": g["grade"],
            "attemptnumber": g.get("attemptnumber", -1),
            "addattempt": 0,
            "workflowstate": workflow_state,
            "applytoall": 0,
            "plugindata[assignfeedbackcomments_editor][text]": g.get(
                "feedback", ""
            ),
            "plugindata[assignfeedbackcomments_editor][format]": 1,
        }
        feedback_file = g.get("feedback_file")
        if feedback_file:
            draft_itemid = self.upload_file(Path(feedback_file))
            params["plugindata[files_filemanager]"] = draft_itemid
        self._call("mod_assign_save_grade", **params)

resolve_credentials(cli_url, cli_token, config)

Resolve Moodle URL and token from CLI flags, env vars, or config.

Priority: CLI flag > environment variable > config file.

Source code in src/mograder/transport/moodle_api.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def resolve_credentials(
    cli_url: str | None,
    cli_token: str | None,
    config,
) -> tuple[str, str]:
    """Resolve Moodle URL and token from CLI flags, env vars, or config.

    Priority: CLI flag > environment variable > config file.
    """
    url = (
        cli_url
        or os.environ.get("MOGRADER_MOODLE_URL")
        or getattr(config, "moodle_url", None)
    )
    token = cli_token or os.environ.get("MOGRADER_MOODLE_TOKEN")

    if not url:
        raise click.UsageError(
            "Moodle URL not set. Provide --url, set MOGRADER_MOODLE_URL, "
            "or add url to [moodle] in mograder.toml"
        )
    if not token:
        cached = load_cached_token(url)
        if cached:
            token = cached["token"]
    if not token:
        raise click.UsageError(
            "Moodle token not set. Provide --token, set MOGRADER_MOODLE_TOKEN, "
            "or run 'mograder moodle login'"
        )

    if url.startswith("http://"):
        click.echo("WARNING: using HTTP — consider HTTPS for security", err=True)

    return url, token

find_assignment(client, course_id, name)

Find an assignment by name within a course.

Tries exact match first, then case-insensitive substring. Raises click.UsageError if no match or ambiguous.

Source code in src/mograder/transport/moodle_api.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def find_assignment(
    client: MoodleAPIClient,
    course_id: int,
    name: str,
) -> dict:
    """Find an assignment by name within a course.

    Tries exact match first, then case-insensitive substring.
    Raises click.UsageError if no match or ambiguous.
    """
    assignments = client.get_assignments(course_id)
    if not assignments:
        raise click.UsageError(f"No assignments found for course {course_id}")

    # Try exact match
    exact = [a for a in assignments if a["name"] == name]
    if len(exact) == 1:
        return exact[0]

    # Try numeric ID
    try:
        aid = int(name)
        by_id = [a for a in assignments if a["id"] == aid]
        if len(by_id) == 1:
            return by_id[0]
    except ValueError:
        pass

    # Try case-insensitive substring
    lower = name.lower()
    matches = [a for a in assignments if lower in a["name"].lower()]
    if len(matches) == 1:
        return matches[0]
    if len(matches) > 1:
        names = "\n  ".join(a["name"] for a in matches)
        raise click.UsageError(
            f"Ambiguous assignment name '{name}'. Matches:\n  {names}"
        )

    names = "\n  ".join(a["name"] for a in assignments)
    raise click.UsageError(f"No assignment matching '{name}'. Available:\n  {names}")

sync_assignments(client, course_id, *, include_pattern=None)

Fetch visible assignments from Moodle and return config-ready dicts.

Each dict has keys: name, id, cmid, duedate, files. File URLs use pluginfile.php (browser-accessible); callers using token auth should go through MoodleAPIClient.download_file which converts automatically.

Source code in src/mograder/transport/moodle_api.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def sync_assignments(
    client: MoodleAPIClient,
    course_id: int,
    *,
    include_pattern: str | None = None,
) -> list[dict]:
    """Fetch visible assignments from Moodle and return config-ready dicts.

    Each dict has keys: name, id, cmid, duedate, files.
    File URLs use ``pluginfile.php`` (browser-accessible); callers using
    token auth should go through ``MoodleAPIClient.download_file`` which
    converts automatically.
    """
    import re

    assignments = client.get_assignments(course_id)

    # Fetch visibility from course contents (cmid → visible)
    course_contents = client._call("core_course_get_contents", courseid=course_id)
    visible_cmids = set()
    for section in course_contents:
        for mod in section.get("modules", []):
            if mod.get("modname") == "assign" and mod.get("visible"):
                visible_cmids.add(mod["id"])

    include_re = re.compile(include_pattern) if include_pattern else None

    result = []
    for a in assignments:
        if a["cmid"] not in visible_cmids:
            continue
        if include_re and not include_re.search(a["name"]):
            continue
        # Convert webservice/pluginfile.php URLs to pluginfile.php (browser-accessible)
        files = []
        for att in a.get("introattachments", []):
            file_url = att["fileurl"].replace(
                "/webservice/pluginfile.php/", "/pluginfile.php/"
            )
            files.append({"name": att["filename"], "url": file_url})
        result.append(
            {
                "name": a["name"],
                "id": a["id"],
                "cmid": a["cmid"],
                "duedate": a["duedate"],
                "files": files,
            }
        )
    return result

request_token(url, username, password, service='moodle_mobile_app')

Exchange username/password for a web service token via /login/token.php.

Returns the token string. Raises MoodleAPIError on failure.

Source code in src/mograder/transport/moodle_api.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
def request_token(
    url: str,
    username: str,
    password: str,
    service: str = "moodle_mobile_app",
) -> str:
    """Exchange username/password for a web service token via /login/token.php.

    Returns the token string.  Raises ``MoodleAPIError`` on failure.
    """
    resp = requests.post(
        f"{url.rstrip('/')}/login/token.php",
        data={"username": username, "password": password, "service": service},
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()
    if "token" not in data:
        raise MoodleAPIError(
            data.get("error", "Login failed"),
            error_code=data.get("errorcode"),
        )
    return data["token"]

build_sso_login_url(url, passport, service='moodle_mobile_app')

Build the SSO launch URL for browser-based token acquisition.

The user visits this URL, logs in via SSO, and is redirected to moodlemobile://token=<base64_encoded_data>.

Source code in src/mograder/transport/moodle_api.py 
483
484
485
486
487
488
489
490
491
492
493
494
def build_sso_login_url(
    url: str,
    passport: str,
    service: str = "moodle_mobile_app",
) -> str:
    """Build the SSO launch URL for browser-based token acquisition.

    The user visits this URL, logs in via SSO, and is redirected to
    ``moodlemobile://token=<base64_encoded_data>``.
    """
    base = url.rstrip("/")
    return f"{base}/admin/tool/mobile/launch.php?service={service}&passport={passport}"

extract_token_from_sso_url(url_string)

Extract the web service token from a moodlemobile://token=... redirect URL.

The token payload is base64-encoded as siteid:::token or siteid:::token:::privatetoken. Returns the token (second field).

Raises MoodleAPIError if the URL cannot be parsed.

Source code in src/mograder/transport/moodle_api.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
def extract_token_from_sso_url(url_string: str) -> str:
    """Extract the web service token from a ``moodlemobile://token=...`` redirect URL.

    The token payload is base64-encoded as ``siteid:::token`` or
    ``siteid:::token:::privatetoken``.  Returns the token (second field).

    Raises ``MoodleAPIError`` if the URL cannot be parsed.
    """
    import base64
    from urllib.parse import urlparse, parse_qs

    parsed = urlparse(url_string)
    # Try scheme-based: moodlemobile://token=ENCODED
    encoded = None
    if parsed.scheme == "moodlemobile":
        # The path is "token=..." (no host)
        path_part = parsed.netloc + parsed.path
        if path_part.startswith("token="):
            encoded = path_part[len("token=") :]
        else:
            # Try query string
            qs = parse_qs(parsed.query)
            if "token" in qs:
                encoded = qs["token"][0]

    if not encoded:
        # Try as raw base64 string (user may have pasted just the token part)
        encoded = url_string.strip()

    # Pad base64 if needed
    padding = 4 - len(encoded) % 4
    if padding < 4:
        encoded += "=" * padding

    try:
        decoded = base64.b64decode(encoded).decode("utf-8")
    except Exception:
        raise MoodleAPIError("Could not decode SSO token — check the pasted URL")

    parts = decoded.split(":::")
    if len(parts) < 2:
        raise MoodleAPIError(
            f"Unexpected token format (got {len(parts)} part(s), expected 2+)"
        )
    return parts[1]

load_cached_token(url)

Load a cached token for url.

Source code in src/mograder/transport/moodle_api.py 
544
545
546
def load_cached_token(url: str) -> dict | None:
    """Load a cached token for *url*."""
    return _moodle_cache().load(match_key="url", match_value=url)

save_cached_token(url, token, fullname)

Persist a token to ~/.config/mograder/token.json.

Source code in src/mograder/transport/moodle_api.py 
549
550
551
def save_cached_token(url: str, token: str, fullname: str) -> None:
    """Persist a token to ``~/.config/mograder/token.json``."""
    _moodle_cache().save({"url": url, "token": token, "fullname": fullname})

clear_cached_token()

Remove the cached token file.

Source code in src/mograder/transport/moodle_api.py 
554
555
556
def clear_cached_token() -> None:
    """Remove the cached token file."""
    _moodle_cache().clear()