aaronrene / scooling-lab public
service.py python
173 lines 6.6 KB
Raw
sha256:fc4c9ad652d1fff3dc508cb6ea02ee710ee6dfc4cb3761291d9900b5e029ea8a feat(slice-7): T3 dataset review lifecycle, job queue, prov… Human minor ⚠ breaking 42 days ago
1 """Application service for Scooling Lab training API operations."""
2
3 from __future__ import annotations
4
5 import threading
6 from datetime import datetime
7 from typing import Iterable
8
9 from scooling_lab.contracts import (
10 TrainingJobRequest,
11 TrainingJobStatus,
12 require_artifact_id,
13 require_job_id,
14 )
15 from scooling_lab.dataset_review import (
16 DatasetStore,
17 RejectionReasonCode,
18 validate_review_request,
19 )
20 from scooling_lab.errors import ApiError, ErrorCode
21 from scooling_lab.fake_worker import FakeTrainingWorker
22 from scooling_lab.store import TrainingJobStore
23
24
25 class TrainingApiService:
26 """Implements the T2/T3 API contract over a store, fake worker, and dataset store.
27
28 Dataset approval is checked before any job is created. A ``threading.Semaphore``
29 enforces the ``max_concurrent_running`` bound from the store so that concurrent
30 callers see at most that many jobs in the running state simultaneously.
31 """
32
33 def __init__(
34 self,
35 store: TrainingJobStore,
36 auto_run_worker: bool = True,
37 dataset_store: DatasetStore | None = None,
38 ) -> None:
39 """Create a service with optional synchronous fake-worker completion.
40
41 If ``dataset_store`` is omitted a default store pre-approving the
42 synthetic fixture dataset is used, so existing callers are unaffected.
43 """
44
45 self._store = store
46 self._worker = FakeTrainingWorker(store)
47 self._auto_run_worker = auto_run_worker
48 self._dataset_store = dataset_store if dataset_store is not None else DatasetStore()
49 # Semaphore mirrors the store's max_concurrent_running for thread safety.
50 self._run_semaphore = threading.Semaphore(store._max_concurrent_running)
51
52 # ------------------------------------------------------------------ dataset
53
54 def register_dataset(self, payload: dict[str, object]) -> dict[str, object]:
55 """Register a dataset for the review lifecycle."""
56
57 dataset_id_raw = payload.get("datasetId")
58 if not isinstance(dataset_id_raw, str):
59 raise ApiError(ErrorCode.VALIDATION_ERROR, 400)
60 record = self._dataset_store.register(dataset_id_raw)
61 return record.to_public_dict()
62
63 def submit_dataset_for_review(self, dataset_id: str) -> dict[str, object]:
64 """Advance a registered dataset to pending_review."""
65
66 record = self._dataset_store.submit_for_review(dataset_id)
67 return record.to_public_dict()
68
69 def review_dataset(
70 self, dataset_id: str, payload: dict[str, object]
71 ) -> dict[str, object]:
72 """Apply an approve or reject decision to a pending-review dataset.
73
74 The ``payload`` must contain ``action: "approve" | "reject"`` and, for
75 rejections, a bounded ``reasonCode`` enum value. No free text is
76 accepted or echoed.
77 """
78
79 action, reason_code = validate_review_request(payload)
80 if action == "approve":
81 record = self._dataset_store.approve(dataset_id)
82 else:
83 assert reason_code is not None
84 record = self._dataset_store.reject(dataset_id, reason_code)
85 return record.to_public_dict()
86
87 def get_dataset(self, dataset_id: str) -> dict[str, object]:
88 """Return the current review status for one dataset."""
89
90 return self._dataset_store.get(dataset_id).to_public_dict()
91
92 # ------------------------------------------------------------------- queue
93
94 def get_queue_state(self) -> dict[str, object]:
95 """Return a content-free snapshot of the job queue state."""
96
97 return self._store.queue_state()
98
99 # -------------------------------------------------------------------- jobs
100
101 def create_training_job(self, payload: dict[str, object]) -> dict[str, object]:
102 """Validate, create, and optionally complete a fixture training job.
103
104 Raises ``DATASET_NOT_APPROVED`` (403) if the requested dataset has not
105 passed the review lifecycle.
106 """
107
108 request = TrainingJobRequest.from_mapping(payload)
109 if not self._dataset_store.is_approved(request.dataset_id):
110 raise ApiError(ErrorCode.DATASET_NOT_APPROVED, 403)
111 job = self._store.create(request)
112 if self._auto_run_worker and job.status in {
113 TrainingJobStatus.QUEUED,
114 TrainingJobStatus.RUNNING,
115 }:
116 acquired = self._run_semaphore.acquire(blocking=True, timeout=10)
117 try:
118 if acquired:
119 job = self._worker.run_job(job.id)
120 finally:
121 if acquired:
122 self._run_semaphore.release()
123 return job.to_public_dict()
124
125 def get_training_job(self, job_id: str) -> dict[str, object]:
126 """Return the public status for one training job."""
127
128 require_job_id(job_id)
129 return self._store.evaluate_expiry(job_id).to_public_dict()
130
131 def cancel_training_job(self, job_id: str) -> dict[str, object]:
132 """Cancel a queued or running training job."""
133
134 require_job_id(job_id)
135 return self._store.update_status(
136 job_id, TrainingJobStatus.CANCELLED
137 ).to_public_dict()
138
139 def list_artifacts(self, job_id: str) -> dict[str, object]:
140 """Return placeholder artifacts registered for one job."""
141
142 require_job_id(job_id)
143 self._store.evaluate_expiry(job_id)
144 artifacts = [artifact.to_dict() for artifact in self._store.list_artifacts(job_id)]
145 return {"jobId": job_id, "artifacts": artifacts}
146
147 def get_provenance(self, job_id: str) -> dict[str, object]:
148 """Return the validated provenance record for one completed fixture job.
149
150 Provenance is also returned for expired (tombstone) jobs when the
151 deletion was triggered by retention TTL, not an explicit delete call.
152 """
153
154 require_job_id(job_id)
155 self._store.evaluate_expiry(job_id)
156 return self._store.get_provenance(job_id).to_dict()
157
158 def delete_artifact(self, job_id: str, artifact_id: str) -> dict[str, object]:
159 """Delete an artifact and all derived content-bearing metadata."""
160
161 require_job_id(job_id)
162 require_artifact_id(artifact_id)
163 return self._store.delete_artifact(job_id, artifact_id).to_dict()
164
165 def sweep_expired_artifacts(self, now: datetime | None = None) -> dict[str, object]:
166 """Evaluate all retention policies and delete expired artifacts."""
167
168 return self._store.sweep_expired(now)
169
170 def verify_deleted_artifact_absence(self, hash_values: Iterable[str]) -> bool:
171 """Verify deleted artifact hashes are absent from all store outputs."""
172
173 return self._store.verify_hash_absence(hash_values)
File History 1 commit
sha256:fc4c9ad652d1fff3dc508cb6ea02ee710ee6dfc4cb3761291d9900b5e029ea8a feat(slice-7): T3 dataset review lifecycle, job queue, prov… Human minor ⚠ 42 days ago