aaronrene / scooling-lab public
dataset_review.py python
268 lines 9.0 KB
Raw
sha256:fc4c9ad652d1fff3dc508cb6ea02ee710ee6dfc4cb3761291d9900b5e029ea8a feat(slice-7): T3 dataset review lifecycle, job queue, prov… Human minor ⚠ breaking 41 days ago
1 """Dataset registration and review lifecycle for the Scooling Lab training API.
2
3 Datasets move through a bounded state machine before any job can reference them:
4 registered → pending_review → approved | rejected
5
6 Only approved datasets are eligible for job submission. Rejection is communicated
7 as a machine-readable ``RejectionReasonCode`` enum; no caller-supplied text is ever
8 echoed back in API responses or logs.
9 """
10
11 from __future__ import annotations
12
13 import re
14 import threading
15 from dataclasses import dataclass, field
16 from enum import Enum
17 from types import MappingProxyType
18 from typing import Mapping
19
20 from scooling_lab.errors import ApiError, ErrorCode
21
22
23 DATASET_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{3,96}$")
24 FORBIDDEN_STRING_RE = re.compile(
25 r"(?i)(https?://|file://|ssh://|[;&|`$<>]|\.\./|/\w|[A-Za-z]:\\)"
26 )
27
28 # The fixture dataset is pre-approved so existing tests require no changes.
29 FIXTURE_APPROVED_DATASET_IDS: frozenset[str] = frozenset(
30 {"fixture:synthetic-tiny-v1"}
31 )
32
33
34 class DatasetStatus(str, Enum):
35 """States in the dataset review lifecycle."""
36
37 REGISTERED = "registered"
38 PENDING_REVIEW = "pending_review"
39 APPROVED = "approved"
40 REJECTED = "rejected"
41
42
43 class RejectionReasonCode(str, Enum):
44 """Bounded machine-readable reason codes for dataset rejection.
45
46 Free text is never accepted or echoed. Only one of these values may
47 appear in a review decision.
48 """
49
50 SYNTHETIC_LIMIT = "SYNTHETIC_LIMIT"
51 FORMAT_INVALID = "FORMAT_INVALID"
52 POLICY_VIOLATION = "POLICY_VIOLATION"
53 SCHEMA_MISMATCH = "SCHEMA_MISMATCH"
54 DUPLICATE_SUBMISSION = "DUPLICATE_SUBMISSION"
55
56
57 _DATASET_TRANSITIONS: MappingProxyType[
58 DatasetStatus, frozenset[DatasetStatus]
59 ] = MappingProxyType(
60 {
61 DatasetStatus.REGISTERED: frozenset({DatasetStatus.PENDING_REVIEW}),
62 DatasetStatus.PENDING_REVIEW: frozenset(
63 {DatasetStatus.APPROVED, DatasetStatus.REJECTED}
64 ),
65 DatasetStatus.APPROVED: frozenset(),
66 DatasetStatus.REJECTED: frozenset(),
67 }
68 )
69
70
71 def dataset_transition(
72 current: DatasetStatus, target: DatasetStatus
73 ) -> DatasetStatus:
74 """Apply the dataset review state machine or raise for invalid moves.
75
76 Idempotent same-state replay is allowed for every state.
77 """
78
79 if current == target:
80 return current
81 if target in _DATASET_TRANSITIONS[current]:
82 return target
83 raise ApiError(ErrorCode.INVALID_TRANSITION, 409)
84
85
86 def require_dataset_id(value: object) -> str:
87 """Validate a dataset id against the safe-identifier contract."""
88
89 if not isinstance(value, str) or not DATASET_ID_RE.fullmatch(value):
90 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
91 if FORBIDDEN_STRING_RE.search(value):
92 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
93 return value
94
95
96 def require_rejection_reason(value: object) -> RejectionReasonCode:
97 """Validate a rejection reason code; reject unknown or free-text values."""
98
99 if not isinstance(value, str):
100 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
101 try:
102 return RejectionReasonCode(value)
103 except ValueError as exc:
104 raise ApiError(ErrorCode.VALIDATION_ERROR, 400) from exc
105
106
107 def _utc_now_iso() -> str:
108 """Return a UTC timestamp formatted for public metadata."""
109
110 from datetime import UTC, datetime
111
112 return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
113
114
115 @dataclass
116 class DatasetRecord:
117 """Internal record for one registered fixture dataset."""
118
119 dataset_id: str
120 status: DatasetStatus = DatasetStatus.REGISTERED
121 registered_at: str = field(default_factory=_utc_now_iso)
122 updated_at: str = field(default_factory=_utc_now_iso)
123 rejection_reason: RejectionReasonCode | None = None
124
125 def to_public_dict(self) -> dict[str, object]:
126 """Serialize the safe dataset status shape for API responses."""
127
128 payload: dict[str, object] = {
129 "datasetId": self.dataset_id,
130 "status": self.status.value,
131 "registeredAt": self.registered_at,
132 "updatedAt": self.updated_at,
133 }
134 if self.rejection_reason is not None:
135 payload["rejectionReasonCode"] = self.rejection_reason.value
136 return payload
137
138
139 class DatasetStore:
140 """Thread-safe, in-memory store for dataset review lifecycle records.
141
142 The store is pre-populated with the fixture dataset IDs from
143 ``FIXTURE_APPROVED_DATASET_IDS`` so that existing job submission flows
144 continue to work without an explicit registration step.
145 """
146
147 def __init__(self) -> None:
148 """Create a store pre-approving all fixture dataset ids."""
149
150 self._lock = threading.RLock()
151 self._datasets: dict[str, DatasetRecord] = {}
152 ts = _utc_now_iso()
153 for dataset_id in FIXTURE_APPROVED_DATASET_IDS:
154 self._datasets[dataset_id] = DatasetRecord(
155 dataset_id=dataset_id,
156 status=DatasetStatus.APPROVED,
157 registered_at=ts,
158 updated_at=ts,
159 )
160
161 def register(self, dataset_id: str) -> DatasetRecord:
162 """Register a dataset; idempotent if already registered.
163
164 Raises ``CONFLICT`` (409) if the same id has already been approved or
165 rejected so that callers cannot silently re-enter a terminal-path
166 dataset into review.
167 """
168
169 require_dataset_id(dataset_id)
170 with self._lock:
171 existing = self._datasets.get(dataset_id)
172 if existing is not None:
173 if existing.status in {DatasetStatus.APPROVED, DatasetStatus.REJECTED}:
174 raise ApiError(ErrorCode.CONFLICT, 409)
175 return existing
176 record = DatasetRecord(dataset_id=dataset_id)
177 self._datasets[dataset_id] = record
178 return record
179
180 def submit_for_review(self, dataset_id: str) -> DatasetRecord:
181 """Advance a registered dataset to pending_review."""
182
183 require_dataset_id(dataset_id)
184 with self._lock:
185 record = self._get_locked(dataset_id)
186 record.status = dataset_transition(
187 record.status, DatasetStatus.PENDING_REVIEW
188 )
189 record.updated_at = _utc_now_iso()
190 return record
191
192 def approve(self, dataset_id: str) -> DatasetRecord:
193 """Approve a dataset that is pending review."""
194
195 require_dataset_id(dataset_id)
196 with self._lock:
197 record = self._get_locked(dataset_id)
198 record.status = dataset_transition(record.status, DatasetStatus.APPROVED)
199 record.updated_at = _utc_now_iso()
200 return record
201
202 def reject(
203 self, dataset_id: str, reason_code: RejectionReasonCode
204 ) -> DatasetRecord:
205 """Reject a dataset that is pending review with a bounded reason code."""
206
207 require_dataset_id(dataset_id)
208 with self._lock:
209 record = self._get_locked(dataset_id)
210 record.status = dataset_transition(record.status, DatasetStatus.REJECTED)
211 record.rejection_reason = reason_code
212 record.updated_at = _utc_now_iso()
213 return record
214
215 def get(self, dataset_id: str) -> DatasetRecord:
216 """Return one dataset record or raise a safe not-found error."""
217
218 require_dataset_id(dataset_id)
219 with self._lock:
220 return self._get_locked(dataset_id)
221
222 def is_approved(self, dataset_id: str) -> bool:
223 """Return True only if the dataset is in the approved state."""
224
225 try:
226 require_dataset_id(dataset_id)
227 except ApiError:
228 return False
229 with self._lock:
230 record = self._datasets.get(dataset_id)
231 return record is not None and record.status == DatasetStatus.APPROVED
232
233 def _get_locked(self, dataset_id: str) -> DatasetRecord:
234 record = self._datasets.get(dataset_id)
235 if record is None:
236 raise ApiError(ErrorCode.NOT_FOUND, 404)
237 return record
238
239
240 def validate_review_request(payload: Mapping[str, object]) -> tuple[str, RejectionReasonCode | None]:
241 """Parse and validate a dataset review action payload.
242
243 Returns ``(action, reason_code)`` where ``reason_code`` is non-None only
244 when ``action == "reject"``. No caller-supplied text is echoed.
245 """
246
247 allowed_keys = {"action", "reasonCode"}
248 if not isinstance(payload, Mapping):
249 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
250 unknown = set(payload) - allowed_keys
251 if unknown:
252 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
253
254 action = payload.get("action")
255 if action not in {"approve", "reject"}:
256 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
257
258 reason_code: RejectionReasonCode | None = None
259 if action == "reject":
260 raw_reason = payload.get("reasonCode")
261 if raw_reason is None:
262 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
263 reason_code = require_rejection_reason(raw_reason)
264 else:
265 if "reasonCode" in payload:
266 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
267
268 return str(action), reason_code
File History 1 commit
sha256:fc4c9ad652d1fff3dc508cb6ea02ee710ee6dfc4cb3761291d9900b5e029ea8a feat(slice-7): T3 dataset review lifecycle, job queue, prov… Human minor ⚠ 41 days ago