diff --git a/CHANGES.md b/CHANGES.md index fa9dcd3d..559f3dbc 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,12 @@ ## Changes in version 0.2.1 (in development) +### Enhancements + +- **Wraptile** now supports restarting failed or dismissed jobs with + `POST /jobs/{jobId}/restart`. Local services reuse the original process + request, while Airflow services replay the original DAG-run configuration. + (#41) + ### Fixes - Fixed `client.show_app()` forcing a redundant interactive OAuth2/PKCE login diff --git a/gavicore/src/gavicore/service/service.py b/gavicore/src/gavicore/service/service.py index 8646f4f8..1834688a 100644 --- a/gavicore/src/gavicore/service/service.py +++ b/gavicore/src/gavicore/service/service.py @@ -1,6 +1,6 @@ # generated by gen_server.py: # filename: service.py: -# timestamp: 2026-05-27T07:55:14.570257 +# timestamp: 2026-08-03T11:32:13.433039 from abc import ABC, abstractmethod @@ -97,3 +97,9 @@ async def get_job_results(self, job_id: str, *args, **kwargs) -> JobResults: For more information, see [OGC API — Processes — Part 1 Section 7.13](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_retrieve_job_results). """ + + @abstractmethod + async def restart_job(self, job_id: str, *args, **kwargs) -> JobInfo: + """Create a new job using the original process request of a failed or + dismissed job. + """ diff --git a/gavicore/tests/test_service.py b/gavicore/tests/test_service.py index ba552952..68065bd0 100644 --- a/gavicore/tests/test_service.py +++ b/gavicore/tests/test_service.py @@ -18,6 +18,7 @@ "get_processes", "get_job", "get_jobs", + "restart_job", } diff --git a/tools/gen_server.py b/tools/gen_server.py index 7319bbd0..349b734b 100644 --- a/tools/gen_server.py +++ b/tools/gen_server.py @@ -55,7 +55,7 @@ def main(): [ "from abc import ABC, abstractmethod\n", "\n", - f"from .models import {model_list}\n", + f"from gavicore.models import {model_list}\n", "\n", "class Service(ABC):\n", service_code, diff --git a/tools/openapi.yaml b/tools/openapi.yaml index c652406d..ef2e5655 100644 --- a/tools/openapi.yaml +++ b/tools/openapi.yaml @@ -382,6 +382,50 @@ paths: text/html: schema: type: string + /jobs/{jobId}/restart: + post: + tags: + - Restart + summary: restart a failed or dismissed job + description: | + Create a new job using the original process request of a failed or + dismissed job. + operationId: restartJob + parameters: + - name: jobId + in: path + description: Local identifier of the job to restart + required: true + schema: + type: string + responses: + "201": + description: The restarted job. + content: + application/json: + schema: + $ref: '#/components/schemas/JobInfo' + "403": + description: The job cannot be restarted. + content: + application/json: + schema: + $ref: '#/components/schemas/ApiError' + "404": + description: The requested job was not found. + content: + application/json: + schema: + $ref: '#/components/schemas/ApiError' + "500": + description: A server error occurred. + content: + application/json: + schema: + $ref: '#/components/schemas/ApiError' + text/html: + schema: + type: string components: schemas: #------------------------------------------------------ diff --git a/wraptile/src/wraptile/routes.py b/wraptile/src/wraptile/routes.py index da47adef..0304873b 100644 --- a/wraptile/src/wraptile/routes.py +++ b/wraptile/src/wraptile/routes.py @@ -1,6 +1,6 @@ # generated by gen_server.py: # filename: routes.py: -# timestamp: 2026-05-27T07:55:14.099104 +# timestamp: 2026-07-31T14:05:53.544409 import fastapi @@ -169,3 +169,20 @@ async def get_job_results( return await service.get_job_results( job_id=jobId, request=request, response=response ) + + +# noinspection PyPep8Naming +@app.post( + "/jobs/{jobId}/restart", + status_code=201, + response_model=JobInfo, + response_model_exclude_none=True, + response_model_exclude_unset=True, +) +async def restart_job( + jobId: str, + request: fastapi.Request, + response: fastapi.Response, + service: Service = fastapi.Depends(get_service), # noqa B008 +): + return await service.restart_job(job_id=jobId, request=request, response=response) diff --git a/wraptile/src/wraptile/services/airflow/airflow_service.py b/wraptile/src/wraptile/services/airflow/airflow_service.py index aa8f7070..8418d065 100644 --- a/wraptile/src/wraptile/services/airflow/airflow_service.py +++ b/wraptile/src/wraptile/services/airflow/airflow_service.py @@ -230,6 +230,25 @@ async def dismiss_job(self, job_id: str, *args, **kwargs) -> JobInfo: ) from e return self.dag_run_to_job_info(dag_run) + async def restart_job(self, job_id: str, *args, **kwargs) -> JobInfo: + """Create a new DAG run using the original run configuration.""" + dag_id = self.get_dag_id_from_job_id(job_id) + try: + dag_run = self.airflow_dag_run_api.get_dag_run(dag_id, job_id) + except ApiException as e: + raise ServiceException( + e.status, e.reason, exception=e, is_job_problem=True + ) from e + job_info = self.dag_run_to_job_info(dag_run) + if job_info.status not in (JobStatus.failed, JobStatus.dismissed): + raise ServiceException( + 403, + detail=f"Job {job_id!r} cannot be restarted unless it failed or was dismissed", + is_job_problem=True, + ) + process_request = ProcessRequest(inputs=dag_run.conf or {}) + return await self.execute_process(dag_id, process_request) + async def get_job_results(self, job_id: str, *args, **kwargs) -> JobResults: dag_id = self.get_dag_id_from_job_id(job_id) return_value: Optional[Any] = None diff --git a/wraptile/src/wraptile/services/local/local_service.py b/wraptile/src/wraptile/services/local/local_service.py index a1c61aa7..06485ebf 100644 --- a/wraptile/src/wraptile/services/local/local_service.py +++ b/wraptile/src/wraptile/services/local/local_service.py @@ -40,6 +40,7 @@ def __init__( self.process_registry = process_registry or ProcessRegistry() self.jobs: dict[str, Job] = {} + self.job_requests: dict[str, ProcessRequest] = {} self.job_results: dict[str, JobResults | None] = {} self.job_uses_processes: dict[str, bool] = {} self._executor_uses_processes = False @@ -117,6 +118,7 @@ async def execute_process( ), ) self.jobs[job_id] = job + self.job_requests[job_id] = process_request.model_copy(deep=True) self.job_uses_processes[job_id] = use_processes if use_processes: assert self.service_ref is not None @@ -160,10 +162,35 @@ async def dismiss_job(self, job_id: str, *args, **_kwargs) -> JobInfo: JobStatus.failed, ): del self.jobs[job_id] + self.job_requests.pop(job_id, None) self.job_results.pop(job_id, None) self.job_uses_processes.pop(job_id, None) return job.job_info + async def restart_job(self, job_id: str, *args, **_kwargs) -> JobInfo: + """Create a new job using a failed or dismissed job's request.""" + job = self._get_job( + job_id, + forbidden_status_codes={ + JobStatus.accepted: "cannot be restarted before it has finished", + JobStatus.running: "cannot be restarted while it is running", + JobStatus.successful: "cannot be restarted after succeeding", + }, + ) + assert job.job_info.status in (JobStatus.failed, JobStatus.dismissed) + process_id = job.job_info.processID + process_request = self.job_requests.get(job_id) + if process_id is None or process_request is None: + raise ServiceException( + 500, + detail=f"Original request for job {job_id!r} is not available", + is_job_problem=True, + ) + return await self.execute_process( + process_id, + process_request.model_copy(deep=True), + ) + async def get_job_results(self, job_id: str, *args, **_kwargs) -> JobResults: job = self._get_job( job_id, diff --git a/wraptile/tests/services/base/test_service_base.py b/wraptile/tests/services/base/test_service_base.py index bf0c8c13..830e0df0 100644 --- a/wraptile/tests/services/base/test_service_base.py +++ b/wraptile/tests/services/base/test_service_base.py @@ -51,6 +51,9 @@ async def get_job(self, job_id: str, *args, **kwargs) -> JobInfo: async def dismiss_job(self, job_id: str, *args, **kwargs) -> JobInfo: raise NotImplementedError + async def restart_job(self, job_id: str, *args, **kwargs) -> JobInfo: + raise NotImplementedError + async def get_job_results(self, job_id: str, *args, **kwargs) -> JobResults: raise NotImplementedError diff --git a/wraptile/tests/services/local/test_local_service.py b/wraptile/tests/services/local/test_local_service.py index a92eaacd..9842817e 100644 --- a/wraptile/tests/services/local/test_local_service.py +++ b/wraptile/tests/services/local/test_local_service.py @@ -266,9 +266,59 @@ async def test_dismiss_finished_job_removes_cached_state(self): self.service.job_uses_processes[job_id] = False await self.service.dismiss_job(job_id=job_id, request=self.get_request()) self.assertNotIn(job_id, self.service.jobs) + self.assertNotIn(job_id, self.service.job_requests) self.assertNotIn(job_id, self.service.job_results) self.assertNotIn(job_id, self.service.job_uses_processes) + async def test_restart_failed_job_reuses_original_request(self): + request = ProcessRequest(inputs={"max_val": 20}) + failed_job = await self.service.execute_process( + process_id="primes_between", + process_request=request, + request=self.get_request(), + ) + self.service.jobs[failed_job.jobID].job_info.status = JobStatus.failed + + restarted_job = await self.service.restart_job( + job_id=failed_job.jobID, + request=self.get_request(), + ) + + self.assertNotEqual(failed_job.jobID, restarted_job.jobID) + self.assertEqual("primes_between", restarted_job.processID) + self.assertEqual(request, self.service.job_requests[restarted_job.jobID]) + + async def test_restart_non_failed_job_fails(self): + job_info = await self.service.execute_process( + process_id="primes_between", + process_request=ProcessRequest(inputs={"max_val": 20}), + request=self.get_request(), + ) + self.service.jobs[job_info.jobID].job_info.status = JobStatus.successful + + with pytest.raises( + ServiceException, match="cannot be restarted after succeeding" + ): + await self.service.restart_job( + job_id=job_info.jobID, + request=self.get_request(), + ) + + async def test_restart_without_original_request_fails(self): + job_info = await self.service.execute_process( + process_id="primes_between", + process_request=ProcessRequest(inputs={"max_val": 20}), + request=self.get_request(), + ) + self.service.jobs[job_info.jobID].job_info.status = JobStatus.failed + del self.service.job_requests[job_info.jobID] + + with pytest.raises(ServiceException, match="Original request.*not available"): + await self.service.restart_job( + job_id=job_info.jobID, + request=self.get_request(), + ) + def test_ensure_executor_reconfigures_missing_executor(self): self.service.executor = None executor = self.service._ensure_executor() diff --git a/wraptile/tests/test_app.py b/wraptile/tests/test_app.py index 1ebe46e8..770032d0 100644 --- a/wraptile/tests/test_app.py +++ b/wraptile/tests/test_app.py @@ -7,6 +7,7 @@ from fastapi.testclient import TestClient +from gavicore.models import JobStatus from wraptile.logging import LogMessageFilter from wraptile.main import app from wraptile.provider import ServiceProvider @@ -71,6 +72,16 @@ def test_dismiss_job(self): response = client.delete(f"/jobs/{job_id}") self.assertEqual(200, response.status_code) + def test_restart_failed_job(self): + response = client.post("/processes/primes_between/execution", json={}) + job_id = response.json()["jobID"] + service.jobs[job_id].job_info.status = JobStatus.failed + + response = client.post(f"/jobs/{job_id}/restart") + + self.assertEqual(201, response.status_code) + self.assertNotEqual(job_id, response.json()["jobID"]) + def test_get_job_results(self): response = client.post("/processes/primes_between/execution", json={}) job_info = response.json()