Spaces:
Sleeping
Sleeping
File size: 5,968 Bytes
476893a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 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 | """Typed client for the Unit 4 GAIA scoring API.
Four routes are exposed by the course backend:
* ``GET /questions`` -> list of question objects
* ``GET /random-question`` -> a single question object
* ``GET /files/{task_id}`` -> the raw file attached to a task (bytes)
* ``POST /submit`` -> score the submitted answers
Every function here wraps one of those routes with timeouts, retry on
transient failures, and clear exceptions. JSON routes return parsed Python
objects; the file route returns raw ``bytes``.
"""
from __future__ import annotations
import time
from typing import Any, Dict, List, Mapping, Sequence
import requests
from . import config
class ScoringAPIError(RuntimeError):
"""Raised when the scoring API returns an error or is unreachable."""
def __init__(self, message: str, *, status_code: int | None = None) -> None:
super().__init__(message)
self.status_code = status_code
# Status codes worth retrying: transient server / gateway errors.
_RETRYABLE_STATUS = {500, 502, 503, 504}
def _request(
method: str,
path: str,
*,
json_body: Any | None = None,
expect_json: bool = True,
) -> Any:
"""Perform an HTTP request with timeout, retry and uniform error handling.
Returns parsed JSON when ``expect_json`` is True, otherwise raw bytes.
Raises :class:`ScoringAPIError` on any unrecoverable failure.
"""
url = f"{config.SCORING_API_BASE}/{path.lstrip('/')}"
last_exc: Exception | None = None
for attempt in range(1, config.MAX_RETRIES + 1):
try:
response = requests.request(
method,
url,
json=json_body,
timeout=config.DEFAULT_TIMEOUT,
)
except requests.RequestException as exc:
# Network-level failure (DNS, connection reset, timeout). Retry, but
# don't sleep after the final attempt — we're about to give up.
last_exc = exc
if attempt < config.MAX_RETRIES:
_backoff(attempt)
continue
# Retry transient server errors; surface everything else immediately.
if response.status_code in _RETRYABLE_STATUS and attempt < config.MAX_RETRIES:
last_exc = ScoringAPIError(
f"{method} {url} returned {response.status_code}",
status_code=response.status_code,
)
_backoff(attempt)
continue
if not response.ok:
raise ScoringAPIError(
f"{method} {url} failed: {response.status_code} {response.text[:300]}",
status_code=response.status_code,
)
if expect_json:
try:
return response.json()
except ValueError as exc:
raise ScoringAPIError(
f"{method} {url} returned non-JSON body: {response.text[:200]}"
) from exc
return response.content
# Exhausted retries.
raise ScoringAPIError(
f"{method} {url} failed after {config.MAX_RETRIES} attempts: {last_exc}"
) from last_exc
def _backoff(attempt: int) -> None:
"""Exponential backoff: 1s, 2s, 4s, ... capped at 16s."""
time.sleep(min(2 ** (attempt - 1), 16))
def get_questions() -> List[Dict[str, Any]]:
"""Return the full list of Unit 4 questions.
Each item is a dict that includes at least ``task_id`` and ``question``;
tasks with attachments also carry a non-empty ``file_name``.
"""
return _request("GET", "/questions")
def get_random_question() -> Dict[str, Any]:
"""Return a single random question dict (handy for quick dry runs)."""
return _request("GET", "/random-question")
def get_file(task_id: str) -> bytes:
"""Download the raw file attached to ``task_id`` and return its bytes.
The caller is responsible for writing the bytes to disk and interpreting
them (csv/xlsx/image/audio/...). See ``files.py`` for that logic.
"""
if not task_id:
raise ValueError("task_id is required to fetch a file")
return _request("GET", f"/files/{task_id}", expect_json=False)
def submit_answers(
username: str,
agent_code: str,
answers: Sequence[Mapping[str, str]],
) -> Dict[str, Any]:
"""Submit answers for scoring and return the parsed result.
Parameters
----------
username:
Your Hugging Face username.
agent_code:
Public Space code link ending in ``/tree/main``.
answers:
Sequence of ``{"task_id": ..., "submitted_answer": ...}`` mappings.
Use :func:`build_answer` / :func:`build_answers` to construct these.
The returned dict typically contains ``score``, ``correct_count``,
``total_attempted`` and a human-readable ``message``.
"""
if not username:
raise ValueError("username is required")
if not agent_code or not agent_code.rstrip("/").endswith("/tree/main"):
raise ValueError("agent_code must be a public Space link ending in /tree/main")
if not answers:
raise ValueError("answers must contain at least one item")
payload = {
"username": username.strip(),
"agent_code": agent_code.strip(),
"answers": list(answers),
}
return _request("POST", "/submit", json_body=payload)
def build_answer(task_id: str, submitted_answer: Any) -> Dict[str, str]:
"""Build a single answer entry in the exact format the API expects.
The submitted answer is coerced to ``str`` because scoring is exact-match
over text; normalization of the *content* belongs in ``normalize.py``.
"""
return {"task_id": str(task_id), "submitted_answer": str(submitted_answer)}
def build_answers(pairs: Mapping[str, Any]) -> List[Dict[str, str]]:
"""Build the full answers list from a ``{task_id: answer}`` mapping."""
return [build_answer(task_id, answer) for task_id, answer in pairs.items()]
|