API Reference

class Judge0Client:
    """Asynchronous client for the Judge0 API."""

    base_url: str
    timeout: float | httpx.Timeout
    _auth_header: str
    _auth_token: SecretStr | None
    _verify_certs: bool
    _client: httpx.AsyncClient | None

    def __init__(
            self,
            base_url: str,
            timeout: float | httpx.Timeout = 10.0,
            auth_header: str = "X-Auth-Token",
            auth_token: str | SecretStr | None = None,
            verify_certs: bool = True,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self._auth_header = auth_header
        if isinstance(auth_token, SecretStr):
            self._auth_token = auth_token
        else:
            self._auth_token = SecretStr(auth_token) if auth_token else None
        self._verify_certs = verify_certs
        self._client = None

    async def __aenter__(self) -> Self:
        self.open()
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:  # noqa: ANN001
        await self.aclose()

    def open(self) -> None:
        """Initialize the underlying httpx client."""
        headers: dict[str, str] = ({
            "Accept": "application/json",
            "Content-Type": "application/json",
        }) | ({
                  self._auth_header: self._auth_token.get_secret_value()
              } if self._auth_token else {})
        if self._client is None:
            self._client = httpx.AsyncClient(
                base_url=self.base_url,
                timeout=self.timeout,
                headers=headers,
                verify=self._verify_certs,
            )

    async def aclose(self) -> None:
        if self._client:
            await self._client.aclose()
            self._client = None

    @property
    def client(self) -> httpx.AsyncClient:
        if self._client is None:
            raise RuntimeError("Client is not initialized. Use 'async with' or call open().")
        return self._client

    async def create_submission(
            self,
            request: BaseSubmission,
    ) -> SubmissionResponse:
        """Create a submission."""

        resp = await self.client.post(
            "/submissions",
            params={"base64_encoded": "true"},
            json=request.to_body(),
        )
        raise_for_status(resp)
        return SubmissionResponse.from_response(resp)

    async def get_submission(
            self,
            token: str,
    ) -> SubmissionDetail:
        """Get submission result by token."""

        resp = await self.client.get(
            url=f"/submissions/{token}",
            params={"base64_encoded": "true"}
        )
        raise_for_status(resp)
        return SubmissionDetail.from_response(resp).decode_base64()

    async def get_workers(self) -> list[WorkersResponse]:
        """Health-check endpoint: returns workers/queues state."""
        resp = await self.client.get("/workers")
        raise_for_status(resp)
        return WorkersResponse.from_response_list(resp)

    async def get_about(self) -> AboutResponse:
        """Returns general information."""
        resp = await self.client.get("/about")
        raise_for_status(resp)
        return AboutResponse.from_response(resp)

    async def get_isolate(self) -> str:
        """Returns result of isolate --version."""
        resp = await self.client.get("/isolate")
        raise_for_status(resp)
        return resp.text

    async def get_license(self) -> str:
        """Returns a license."""
        resp = await self.client.get("/license")
        raise_for_status(resp)
        return resp.text

class Judge0Error(Exception):
    """Generic Judge0 client exception."""

class BaseSubmission(BaseModel, ABC):
    """Request model for creating a submission in Judge0."""

    compiler_options: str | None = Field(
        default=None, description="Options for the compiler (i.e. compiler flags)", max_length=512
    )
    command_line_arguments: str | None = Field(
        default=None, description="Command line arguments for the program", max_length=512
    )
    stdin: str | None = Field(default=None, description="Input for program")
    expected_output: str | None = Field(
        default=None, description="Expected output of program. Used when you want to compare with stdout"
    )
    cpu_time_limit: float | None = Field(
        default=None, description="Default runtime limit for every program (seconds)"
    )
    cpu_extra_time: float | None = Field(
        default=None,
        description="When a time limit is exceeded, wait for extra time, before killing the program (seconds)"
    )
    wall_time_limit: float | None = Field(
        default=None, description="Limit wall-clock time in seconds (seconds)"
    )
    memory_limit: float | None = Field(default=None, description="Limit address space of the program (kilobytes)")
    stack_limit: int | None = Field(default=None, description="Limit process stack (kilobytes)")
    max_processes_and_or_threads: int | None = Field(
        default=None, description="Maximum number of processes and/or threads program can create"
    )
    enable_per_process_and_thread_time_limit: bool | None = Field(
        default=None, description="If true then cpu_time_limit will be used as per process and thread"
    )
    enable_per_process_and_thread_memory_limit: bool | None = Field(
        default=None, description="If true then memory_limit will be used as per process and thread"
    )
    max_file_size: int | None = Field(
        default=None, description="Limit file size created or modified by the program (kilobytes)"
    )
    redirect_stderr_to_stdout: bool | None = Field(
        default=None, description="If true standard error will be redirected to standard output"
    )
    enable_network: bool | None = Field(default=None, description="	If true program will have network access")
    number_of_runs: int | None = Field(
        default=None, description="Run each program number_of_runs times and take average of time and memory"
    )
    callback_url: str | None = Field(
        default=None,
        description="URL on which Judge0 will issue PUT request with the submission in a request body after submission has been done"
    )

    @abstractmethod
    def to_body(self) -> dict[str, Any]:
        pass

class SingleFileSubmission(BaseSubmission, BaseModel):
    """Request model for creating a single-file submission in Judge0."""

    source_code: str = Field(description="Program’s source code")
    language_id: int = Field(description="Programming language ID")
    additional_files: Mapping[str, str | bytes] | None = Field(
        default=None,
        description="Additional files that should be available alongside the source code (encoded zip)"
    )

    def to_body(self) -> dict[str, Any]:
        data = self.model_dump(exclude_none=True)
        fields_to_encode = ["source_code", "stdin", "expected_output"]
        for f in fields_to_encode:
            if f in data and data[f] is not None:
                data[f] = base64_encode(data[f])
        if "additional_files" in data:
            data["additional_files"] = create_encoded_zip(data["additional_files"])
        return data

class MultiFileSubmission(BaseSubmission, BaseModel):
    """Request model for creating a multi-file submission in Judge0."""

    language_id: Literal[89] = Field(default=89, description="Programming language ID")
    additional_files: Mapping[str, str | bytes] = Field(
        description="Scripts to run and compile and additional files"
    )

    def to_body(self) -> dict[str, Any]:
        data = self.model_dump(exclude_none=True)
        fields_to_encode = ["stdin", "expected_output"]
        for f in fields_to_encode:
            if f in data and data[f] is not None:
                data[f] = base64_encode(data[f])
        if "additional_files" in data:
            data["additional_files"] = create_encoded_zip(data["additional_files"])
        return data

class SubmissionResponse(BaseResponseModel):
    """Response for non-waiting submission — contains only a token."""

    token: str

class SubmissionDetail(BaseResponseModel):
    """Detailed information about submission execution."""

    token: str
    status: SubmissionStatus
    stdout: str | None = None
    stderr: str | None = None
    compile_output: str | None = None
    message: str | None = None
    time: float | None = None
    memory: int | None = None
    exit_code: int | None = None
    language_id: int | None = None

    def decode_base64(self) -> Self:
        return self.model_copy(update={
            "stdout": base64_decode(self.stdout) if self.stdout else self.stdout,
            "stderr": base64_decode(self.stderr) if self.stderr else self.stderr,
            "compile_output": base64_decode(self.compile_output) if self.compile_output else self.compile_output,
            "message": base64_decode(self.message) if self.message else self.message,
        })

class WorkersResponse(BaseResponseModel):
    queue: str = Field(description="Queue name")
    size: int = Field(description="Queue size, number of submissions that are currently waiting to be processed")
    available: int = Field(description="Available number of workers")
    idle: int = Field(description="How many workers are idle")
    working: int = Field(description="How many workers are currently working")
    paused: int = Field(description="How many workers are paused")
    failed: int = Field(description="How many jobs failed")