aaronrene / scooling-lab public
api.py python
213 lines 8.2 KB
Raw
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ breaking 41 days ago
1 """Dependency-free HTTP API for the Scooling Lab T2/T3 contract."""
2
3 from __future__ import annotations
4
5 import json
6 from http import HTTPStatus
7 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
8 from pathlib import Path
9 import re
10 from typing import Callable
11 from urllib.parse import urlparse
12
13 from scooling_lab.errors import ApiError, ErrorCode, error_payload
14 from scooling_lab.service import TrainingApiService
15 from scooling_lab.store import TrainingJobStore
16
17
18 MAX_BODY_BYTES = 16_384
19 JOB_ID_RE = re.compile(r"^job_[a-f0-9]{24}$")
20 ARTIFACT_ID_RE = re.compile(r"^artifact_[a-f0-9]{24}$")
21 DATASET_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{3,96}$")
22
23
24 def parse_job_route(path: str, suffix: str = "") -> str | None:
25 """Extract a safe job id from supported job subresource routes."""
26
27 prefix = "/training/jobs/"
28 if not path.startswith(prefix):
29 return None
30 remainder = path.removeprefix(prefix)
31 if suffix:
32 ending = f"/{suffix}"
33 if not remainder.endswith(ending):
34 return None
35 remainder = remainder[: -len(ending)]
36 if "/" in remainder or not JOB_ID_RE.fullmatch(remainder):
37 return None
38 return remainder
39
40
41 def parse_artifact_route(path: str) -> tuple[str, str] | None:
42 """Extract safe job and artifact ids from artifact deletion routes."""
43
44 prefix = "/training/jobs/"
45 marker = "/artifacts/"
46 if not path.startswith(prefix) or marker not in path:
47 return None
48 remainder = path.removeprefix(prefix)
49 job_id, separator, artifact_id = remainder.partition(marker)
50 if separator != marker:
51 return None
52 if "/" in artifact_id:
53 return None
54 if not JOB_ID_RE.fullmatch(job_id) or not ARTIFACT_ID_RE.fullmatch(artifact_id):
55 return None
56 return job_id, artifact_id
57
58
59 def parse_dataset_route(path: str, suffix: str = "") -> str | None:
60 """Extract a safe dataset id from supported dataset subresource routes."""
61
62 prefix = "/datasets/"
63 if not path.startswith(prefix):
64 return None
65 remainder = path.removeprefix(prefix)
66 if suffix:
67 ending = f"/{suffix}"
68 if not remainder.endswith(ending):
69 return None
70 remainder = remainder[: -len(ending)]
71 if "/" in remainder or not DATASET_ID_RE.fullmatch(remainder):
72 return None
73 return remainder
74
75
76 def make_handler(service: TrainingApiService) -> type[BaseHTTPRequestHandler]:
77 """Create a request handler bound to the supplied service."""
78
79 class ScoolingLabRequestHandler(BaseHTTPRequestHandler):
80 """HTTP handler exposing job, artifact, queue, and dataset endpoints."""
81
82 server_version = "ScoolingLab/0.1"
83
84 def do_POST(self) -> None:
85 """Handle createTrainingJob, cancelTrainingJob, and dataset routes."""
86
87 path = urlparse(self.path).path
88 if path == "/training/jobs":
89 self._handle_json(lambda: service.create_training_job(self._read_json()))
90 return
91 retry_job_id = parse_job_route(path, "retry")
92 if retry_job_id is not None:
93 self._handle_json(lambda: service.retry_training_job(retry_job_id))
94 return
95 cancel_job_id = parse_job_route(path, "cancel")
96 if cancel_job_id is not None:
97 self._handle_json(lambda: service.cancel_training_job(cancel_job_id))
98 return
99 if path == "/datasets":
100 self._handle_json(lambda: service.register_dataset(self._read_json()))
101 return
102 review_dataset_id = parse_dataset_route(path, "review")
103 if review_dataset_id is not None:
104 _id = review_dataset_id
105 self._handle_json(
106 lambda: service.review_dataset(_id, self._read_json())
107 )
108 return
109 submit_dataset_id = parse_dataset_route(path, "submit")
110 if submit_dataset_id is not None:
111 _sid = submit_dataset_id
112 self._handle_json(lambda: service.submit_dataset_for_review(_sid))
113 return
114 self._send_error(ApiError(ErrorCode.NOT_FOUND, 404))
115
116 def do_GET(self) -> None:
117 """Handle getTrainingJob, listArtifacts, queue state, and dataset routes."""
118
119 path = urlparse(self.path).path
120 artifacts_job_id = parse_job_route(path, "artifacts")
121 if artifacts_job_id is not None:
122 self._handle_json(lambda: service.list_artifacts(artifacts_job_id))
123 return
124 provenance_job_id = parse_job_route(path, "provenance")
125 if provenance_job_id is not None:
126 self._handle_json(lambda: service.get_provenance(provenance_job_id))
127 return
128 job_id = parse_job_route(path)
129 if job_id is not None:
130 self._handle_json(lambda: service.get_training_job(job_id))
131 return
132 if path == "/training/queue":
133 self._handle_json(service.get_queue_state)
134 return
135 dataset_id = parse_dataset_route(path)
136 if dataset_id is not None:
137 _did = dataset_id
138 self._handle_json(lambda: service.get_dataset(_did))
139 return
140 self._send_error(ApiError(ErrorCode.NOT_FOUND, 404))
141
142 def do_PUT(self) -> None:
143 """Reject unsupported mutation routes with a stable error."""
144
145 self._send_error(ApiError(ErrorCode.METHOD_NOT_ALLOWED, 405))
146
147 def do_DELETE(self) -> None:
148 """Handle idempotent deleteArtifact routes."""
149
150 path = urlparse(self.path).path
151 artifact_route = parse_artifact_route(path)
152 if artifact_route is not None:
153 job_id, artifact_id = artifact_route
154 self._handle_json(lambda: service.delete_artifact(job_id, artifact_id))
155 return
156 self._send_error(ApiError(ErrorCode.METHOD_NOT_ALLOWED, 405))
157
158 def log_message(self, format: str, *args: object) -> None:
159 """Suppress default request logging to avoid payload/path leakage."""
160
161 return
162
163 def _read_json(self) -> dict[str, object]:
164 content_length = self.headers.get("Content-Length")
165 if content_length is None:
166 raise ApiError(ErrorCode.MALFORMED_JSON, 400)
167 length = int(content_length)
168 if length <= 0 or length > MAX_BODY_BYTES:
169 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
170 try:
171 payload = json.loads(self.rfile.read(length).decode("utf-8"))
172 except (UnicodeDecodeError, json.JSONDecodeError) as exc:
173 raise ApiError(ErrorCode.MALFORMED_JSON, 400) from exc
174 if not isinstance(payload, dict):
175 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
176 return payload
177
178 def _handle_json(self, action: Callable[[], object]) -> None:
179 try:
180 result = action()
181 except ApiError as error:
182 self._send_error(error)
183 return
184 except Exception:
185 self._send_error(ApiError(ErrorCode.INTERNAL_ERROR, 500))
186 return
187 self._send_json(result, HTTPStatus.OK)
188
189 def _send_error(self, error: ApiError) -> None:
190 self._send_json(error_payload(error), HTTPStatus(error.status))
191
192 def _send_json(self, payload: object, status: HTTPStatus) -> None:
193 body = json.dumps(payload, sort_keys=True).encode("utf-8")
194 self.send_response(status.value)
195 self.send_header("Content-Type", "application/json")
196 self.send_header("Content-Length", str(len(body)))
197 self.send_header("Cache-Control", "no-store")
198 self.end_headers()
199 self.wfile.write(body)
200
201 return ScoolingLabRequestHandler
202
203
204 def run_server(host: str, port: int, persistence_path: Path | None = None) -> None:
205 """Run the Scooling Lab API server until interrupted."""
206
207 service = TrainingApiService(TrainingJobStore(persistence_path=persistence_path))
208 server = ThreadingHTTPServer((host, port), make_handler(service))
209 server.serve_forever()
210
211
212 if __name__ == "__main__":
213 run_server("127.0.0.1", 8080)
File History 1 commit
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ 41 days ago