feat: integrated the retrieval of the results (#6)

Author: JanssenBrm

app/database/models/processing_job.py

@@ -99,3 +99,20 @@ def update_job_status_by_id(
         logger.warning(
             f"Could not update job status of job {job_id} as it could not be found in the database"
         )
+
+
+def update_job_result_by_id(database: Session, job_id: int, result_link: str):
+    logger.info(
+        f"Updating the result link of processing job with ID {job_id} to {result_link}"
+    )
+    job = get_job_by_id(database, job_id)
+
+    if job:
+        job.result_link = result_link
+        database.commit()
+        database.refresh(job)
+    else:
+        logger.warning(
+            f"Could not update job result link of job {job_id} as it could not be found in "
+            "the database"
+        )

app/platforms/base.py

@@ -34,3 +34,14 @@ def get_job_status(
         :return: Return the processing status
         """
         pass
+
+    @abstractmethod
+    def get_job_result_url(self, job_id: str, details: ServiceDetails) -> str:
+        """
+        Retrieve the job results of a processing job that is running on the platform.
+
+        :param job_id: The ID of the job on the platform
+        :param details: The service details containing the service ID and application.
+        :return: URL where the job results are described
+        """
+        pass

app/platforms/implementations/ogc_api_process.py

@@ -24,3 +24,8 @@ def get_job_status(
         raise NotImplementedError(
             "OGC API Process job status retrieval not implemented yet."
         )
+
+    def get_job_result_url(self, job_id: str, details: ServiceDetails) -> str:
+        raise NotImplementedError(
+            "OGC API Process job result retrieval not implemented yet."
+        )

app/platforms/implementations/openeo.py

@@ -161,3 +161,17 @@ def get_job_status(
             raise SystemError(
                 f"Failed to fetch status openEO job with ID {job_id}"
             ) from e
+
+    def get_job_result_url(self, job_id: str, details: ServiceDetails) -> str:
+        try:
+            logger.debug(f"Fetching job result for openEO job with ID {job_id}")
+            connection = self._setup_connection(details.service)
+            job = connection.job(job_id)
+            return f"{details.service}{job.get_results_metadata_url()}"
+        except Exception as e:
+            logger.exception(
+                f"Failed to fetch result url for for openEO job with ID {job_id}"
+            )
+            raise SystemError(
+                f"Failed to fetch result url for openEO job with ID {job_id}"
+            ) from e

app/services/processing.py

@@ -6,6 +6,7 @@
     get_job_by_user_id,
     get_jobs_by_user_id,
     save_job_to_db,
+    update_job_result_by_id,
     update_job_status_by_id,
 )
 from app.platforms.dispatcher import get_processing_platform
@@ -57,6 +58,13 @@ def get_job_status(job: ProcessingJobRecord) -> ProcessingStatusEnum:
     return platform.get_job_status(job.platform_job_id, details)
 
 
+def get_job_result_url(job: ProcessingJobRecord) -> str:
+    logger.info(f"Retrieving job result for job: {job.platform_job_id}")
+    platform = get_processing_platform(job.label)
+    details = ServiceDetails.model_validate_json(job.service_record)
+    return platform.get_job_result_url(job.platform_job_id, details)
+
+
 def get_processing_jobs_by_user_id(
     database: Session, user_id: str
 ) -> List[ProcessingJobSummary]:
@@ -72,6 +80,7 @@ def get_processing_jobs_by_user_id(
     }
 
     for record in records:
+        # Only check status for active jobs
         if record.status not in inactive_statuses:
             latest_status = get_job_status(record)
             if latest_status != record.status:
@@ -80,6 +89,11 @@ def get_processing_jobs_by_user_id(
         else:
             status = record.status
 
+        # Update the result if the job is finished and results weren't retrieved yet
+        if status == ProcessingStatusEnum.FINISHED and not record.result_link:
+            result_link = get_job_result_url(record)
+            update_job_result_by_id(database, record.id, result_link)
+
         jobs.append(
             ProcessingJobSummary(
                 id=record.id, title=record.title, label=record.label, status=status

tests/platforms/test_dispatcher.py

@@ -17,6 +17,9 @@ def execute_job(
     def get_job_status(self, job_id, details):
         return ProcessingStatusEnum.FINISHED
 
+    def get_job_result_url(self, job_id, details):
+        return "https://foo.bar"
+
 
 @pytest.fixture(autouse=True)
 def clear_registry():

tests/platforms/test_openeo_platform.py

@@ -13,6 +13,7 @@ class DummyOpenEOClient:
     def job(self, job_id):
         job = MagicMock()
         job.status.return_value = ProcessingStatusEnum.RUNNING
+        job.get_results_metadata_url.return_value = "/job/results"
         return job
 
 
@@ -140,24 +141,43 @@ def test_map_openeo_status(openeo_status, expected_enum):
     assert result == expected_enum
 
 
-def test_get_job_status_success():
-    platform = OpenEOPlatform()
+@patch.object(OpenEOPlatform, "_setup_connection")
+def test_get_job_status_success(mock_connection, platform):
+    mock_connection.return_value = DummyOpenEOClient()
 
-    with patch.object(platform, "_setup_connection", return_value=DummyOpenEOClient()):
-        details = ServiceDetails(service="foo", application="bar")
-        result = platform.get_job_status("job123", details)
+    details = ServiceDetails(service="foo", application="bar")
+    result = platform.get_job_status("job123", details)
 
-        assert result == ProcessingStatusEnum.RUNNING
+    assert result == ProcessingStatusEnum.RUNNING
 
 
-def test_get_job_status_error():
-    platform = OpenEOPlatform()
+@patch.object(OpenEOPlatform, "_setup_connection")
+def test_get_job_status_error(mock_connection, platform):
+    mock_connection.side_effect = RuntimeError("Connection error")
 
-    with patch.object(
-        platform, "_setup_connection", side_effect=RuntimeError("Connection error")
-    ):
-        details = ServiceDetails(service="foo", application="bar")
-        with pytest.raises(SystemError) as exc_info:
-            platform.get_job_status("job123", details)
+    details = ServiceDetails(service="foo", application="bar")
+    with pytest.raises(SystemError) as exc_info:
+        platform.get_job_status("job123", details)
 
     assert "Failed to fetch status" in str(exc_info.value)
+
+
+@patch.object(OpenEOPlatform, "_setup_connection")
+def test_get_job_result_success(mock_connection, platform):
+    mock_connection.return_value = DummyOpenEOClient()
+
+    details = ServiceDetails(service="foo", application="bar")
+    result = platform.get_job_result_url("job123", details)
+
+    assert result == "foo/job/results"
+
+
+@patch.object(OpenEOPlatform, "_setup_connection")
+def test_get_job_url_error(mock_connection, platform):
+    mock_connection.side_effect = RuntimeError("Connection error")
+
+    details = ServiceDetails(service="foo", application="bar")
+    with pytest.raises(SystemError) as exc_info:
+        platform.get_job_result_url("job123", details)
+
+    assert "Failed to fetch result url" in str(exc_info.value)

tests/services/test_processing.py

@@ -13,6 +13,7 @@
 )
 from app.services.processing import (
     create_processing_job,
+    get_job_result_url,
     get_job_status,
     get_processing_job_by_user_id,
     get_processing_jobs_by_user_id,
@@ -93,7 +94,7 @@ def test_get_processing_jobs_with_active_and_inactive_statuses(
         platform_job_id="platform456",
         label=ProcessTypeEnum.OGC_API_PROCESS,
         title="Finished Job",
-        status=ProcessingStatusEnum.FINISHED,
+        status=ProcessingStatusEnum.FAILED,
         service_record=json.dumps({"foo": "bar"}),
     )
     mock_get_jobs.return_value = [fake_processing_job_record, inactive_job]
@@ -104,7 +105,7 @@ def test_get_processing_jobs_with_active_and_inactive_statuses(
     assert len(results) == 2
     assert isinstance(results[0], ProcessingJobSummary)
     assert results[0].status == ProcessingStatusEnum.RUNNING
-    assert results[1].status == ProcessingStatusEnum.FINISHED
+    assert results[1].status == ProcessingStatusEnum.FAILED
 
     # Active job should be refreshed
     mock_get_job_status.assert_called_once_with(fake_processing_job_record)
@@ -136,6 +137,44 @@ def test_get_processing_jobs_no_updates(
     mock_update_job_status.assert_not_called()
 
 
+@patch("app.services.processing.update_job_result_by_id")
+@patch("app.services.processing.get_job_result_url")
+@patch("app.services.processing.get_jobs_by_user_id")
+def test_get_processing_jobs_with_finished_statuses(
+    mock_get_jobs, mock_get_jobs_results, mock_update_job_result, fake_db_session
+):
+    finished_job_no_result = ProcessingJobRecord(
+        id=2,
+        platform_job_id="platform456",
+        label=ProcessTypeEnum.OGC_API_PROCESS,
+        title="Finished Job",
+        status=ProcessingStatusEnum.FINISHED,
+        service_record=json.dumps({"foo": "bar"}),
+    )
+    finished_job_result = ProcessingJobRecord(
+        id=3,
+        platform_job_id="platform456",
+        label=ProcessTypeEnum.OGC_API_PROCESS,
+        title="Finished Job",
+        status=ProcessingStatusEnum.FINISHED,
+        service_record=json.dumps({"foo": "bar"}),
+        result_link="https://foo.bar",
+    )
+    mock_get_jobs.return_value = [finished_job_no_result, finished_job_result]
+    mock_get_jobs_results.return_value = "https://foo2.bar"
+
+    results = get_processing_jobs_by_user_id(fake_db_session, "user1")
+
+    assert len(results) == 2
+    assert isinstance(results[0], ProcessingJobSummary)
+
+    # Active job should be refreshed
+    mock_get_jobs_results.assert_called_once_with(finished_job_no_result)
+    mock_update_job_result.assert_called_once_with(
+        ANY, finished_job_no_result.id, "https://foo2.bar"
+    )
+
+
 @patch("app.services.processing.get_processing_platform")
 def test_get_job_status_from_platform(mock_get_platform, fake_processing_job_record):
 
@@ -148,6 +187,18 @@ def test_get_job_status_from_platform(mock_get_platform, fake_processing_job_rec
     assert status == ProcessingStatusEnum.QUEUED
 
 
+@patch("app.services.processing.get_processing_platform")
+def test_get_job_result_from_platform(mock_get_platform, fake_processing_job_record):
+
+    fake_platform = MagicMock()
+    fake_platform.get_job_result_url.return_value = "https://foo.bar"
+    mock_get_platform.return_value = fake_platform
+
+    result = get_job_result_url(fake_processing_job_record)
+
+    assert result == "https://foo.bar"
+
+
 @patch("app.services.processing.get_job_by_user_id")
 def test_get_processing_job_by_user_id(mock_get_job, fake_db_session):