Skip to content

Transport Protocol

The Transport protocol defines the interface for assignment distribution backends (Moodle, HTTPS).

mograder.transport.transport

Transport protocol for remote assignment servers.

TransportError

Bases: Exception

Raised when a transport operation fails.

Source code in src/mograder/transport/transport.py 
11
12
class TransportError(Exception):
    """Raised when a transport operation fails."""

Transport

Bases: Protocol

Protocol for fetching/submitting assignments from a remote server.

Source code in src/mograder/transport/transport.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@runtime_checkable
class Transport(Protocol):
    """Protocol for fetching/submitting assignments from a remote server."""

    def list_assignments(self) -> list[RemoteAssignment]: ...

    def download_file(self, url: str, dest: Path) -> Path: ...

    def submit_file(self, assignment: str, filepath: Path) -> None: ...

    def get_submissions(self, assignment: str) -> list[RemoteSubmission]: ...

    def upload_grades(
        self,
        assignment: str,
        grades: list[dict],
        workflow_state: str = "",
    ) -> None: ...

    def get_status(self, assignment: str) -> RemoteStatus: ...

build_transport(config)

Build a Transport from a MograderConfig.

Reads config.transport to decide which implementation to use.

Source code in src/mograder/transport/transport.py 
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
def build_transport(config) -> Transport:
    """Build a Transport from a MograderConfig.

    Reads ``config.transport`` to decide which implementation to use.
    """
    transport_type = getattr(config, "transport", "moodle")

    if transport_type == "https":
        from mograder.core.auth import load_cached_https_token
        from mograder.transport.https_transport import HTTPSTransport

        url = getattr(config, "https_url", None)
        if not url:
            raise TransportError(
                "HTTPS transport selected but no URL configured. "
                "Set [https] url in mograder.toml"
            )
        # Token resolution: config > env var > cache
        token = getattr(config, "https_token", None) or ""
        user = ""
        if not token:
            import os

            token = os.environ.get("MOGRADER_HTTPS_TOKEN", "")
        if not token:
            cached = load_cached_https_token(url)
            if cached:
                token = cached["token"]
                user = cached.get("user", "")
        if token and not user and ":" in token:
            user = token.split(":", 1)[0]
        return HTTPSTransport(url, user=user, token=token)

    if transport_type == "moodle":
        from mograder.transport.moodle_api import MoodleAPIClient, load_cached_token
        from mograder.transport.moodle_transport import MoodleTransport

        url = getattr(config, "moodle_url", None)
        course_id = getattr(config, "moodle_course_id", None)
        if not url or not course_id:
            raise TransportError(
                "Moodle transport selected but URL or course_id not configured. "
                "Set [moodle] url and course_id in mograder.toml"
            )
        cached = load_cached_token(url)
        if not cached:
            raise TransportError(
                "No Moodle token found. Run 'mograder moodle login' first."
            )
        client = MoodleAPIClient(url, cached["token"])
        return MoodleTransport(client, course_id)

    raise TransportError(f"Unknown transport type: {transport_type!r}")