Initial implementation (without submission endpoints) of Sulu clients.

Author: fkdosilovic

examples/sulu_clients.py

@@ -0,0 +1,20 @@
+import os
+
+import judge0
+from dotenv import load_dotenv
+
+load_dotenv()
+
+sulu_auth_token = os.getenv("SULU_API_KEY")
+
+client_ce = judge0.SuluCEClient(auth_token=sulu_auth_token)
+print(client_ce.get_about())
+print(client_ce.get_config_info())
+print(client_ce.get_statuses())
+print(client_ce.get_languages())
+
+client_extra_ce = judge0.SuluExtraCEClient(auth_token=sulu_auth_token)
+print(client_extra_ce.get_about())
+print(client_extra_ce.get_config_info())
+print(client_extra_ce.get_statuses())
+print(client_extra_ce.get_languages())

src/judge0/__init__.py

@@ -0,0 +1,3 @@
+from .clients import SuluCEClient, SuluExtraCEClient
+
+__all__ = [SuluCEClient, SuluExtraCEClient]

src/judge0/base.py

@@ -0,0 +1,21 @@
+from enum import IntEnum
+
+DEFAULT_SULU_CE_ENDPOINT: str = "https://judge0-ce.p.sulu.sh"
+DEFAULT_SULU_EXTRA_CE_ENDPOINT: str = "https://judge0-extra-ce.p.sulu.sh"
+
+
+class Status(IntEnum):
+    IN_QUEUE = 1
+    PROCESSING = 2
+    ACCEPTED = 3
+    WRONG_ANSWER = 4
+    TIME_LIMIT_EXCEEDED = 5
+    COMPILATION_ERROR = 6
+    RUNTIME_ERROR_SIGSEGV = 7
+    RUNTIME_ERROR_SIGXFSZ = 8
+    RUNTIME_ERROR_SIGFPE = 9
+    RUNTIME_ERROR_SIGABRT = 10
+    RUNTIME_ERROR_NZEC = 11
+    RUNTIME_ERROR_OTHER = 12
+    INTERNAL_ERROR = 13
+    EXEC_FORMAT_ERROR = 14

src/judge0/clients.py

@@ -0,0 +1,76 @@
+from abc import abstractmethod
+from typing import Union
+
+import requests
+
+from .base import DEFAULT_SULU_CE_ENDPOINT, DEFAULT_SULU_EXTRA_CE_ENDPOINT
+
+
+class BaseSuluClient:
+
+    def __init__(
+        self,
+        *,
+        endpoint: Union[str, None] = None,
+        auth_token: Union[str, None] = None,
+    ):
+        if endpoint is None:
+            endpoint = self.default_endpoint
+        self.endpoint = endpoint
+        self.auth_token = auth_token
+        self._session = requests.Session()
+
+    def get_about(self) -> dict:
+        # TODO: Potentially think about caching the successful return.
+        headers = {"Authorization": f"Bearer {self.auth_token}"}
+        r = requests.get(f"{self.endpoint}/about", headers=headers)
+        r.raise_for_status()
+        self._session.headers.update(headers)
+        return r.json()
+
+    def get_config_info(self) -> dict:
+        # TODO: Potentially think about caching the successful return.
+        headers = {"Authorization": f"Bearer {self.auth_token}"}
+        r = requests.get(f"{self.endpoint}/config_info", headers=headers)
+        r.raise_for_status()
+        self._session.headers.update(headers)
+        return r.json()
+
+    def get_statuses(self) -> list[dict]:
+        # TODO: Potentially think about caching the successful return.
+        # TODO: Add docs about Status enum.
+        headers = {"Authorization": f"Bearer {self.auth_token}"}
+        r = requests.get(f"{self.endpoint}/statuses", headers=headers)
+        r.raise_for_status()
+        self._session.headers.update(headers)
+        return r.json()
+
+    def get_languages(self, *, language_id: Union[int, None] = None) -> list[dict]:
+        # TODO: Potentially think about caching the successful return.
+        headers = {"Authorization": f"Bearer {self.auth_token}"}
+        request_url = f"{self.endpoint}/languages"
+        if language_id is not None:
+            request_url = f"{request_url}/{language_id}"
+        r = requests.get(request_url, headers=headers)
+        r.raise_for_status()
+        self._session.headers.update(headers)
+        return r.json()
+
+    @property
+    @abstractmethod
+    def default_endpoint(self) -> str:
+        raise NotImplementedError("Subclasses must define a default endpoint property.")
+
+
+class SuluCEClient(BaseSuluClient):
+
+    @property
+    def default_endpoint(self):
+        return DEFAULT_SULU_CE_ENDPOINT
+
+
+class SuluExtraCEClient(BaseSuluClient):
+
+    @property
+    def default_endpoint(self):
+        return DEFAULT_SULU_EXTRA_CE_ENDPOINT