feat: integration of endpoint for retrieving results

Author: JanssenBrm

alembic/versions/8a69e1ee3fef_update_of_result_of_processing_job.py

@@ -0,0 +1,32 @@
+"""update of result of processing job
+
+Revision ID: 8a69e1ee3fef
+Revises: 046ada00e750
+Create Date: 2025-08-25 09:40:54.675672
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '8a69e1ee3fef'
+down_revision: Union[str, Sequence[str], None] = '046ada00e750'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('processing_jobs', 'result_link')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('processing_jobs', sa.Column('result_link', sa.VARCHAR(), autoincrement=False, nullable=True))
+    # ### end Alembic commands ###

app/database/models/processing_job.py

@@ -1,4 +1,5 @@
 import datetime
+import json
 from typing import List, Optional
 
 from loguru import logger
@@ -8,6 +9,8 @@
 from app.database.db import Base
 from app.schemas.unit_job import ProcessingStatusEnum, ProcessTypeEnum
 
+from stac_pydantic import Collection
+
 
 class ProcessingJobRecord(Base):
     __tablename__ = "processing_jobs"
@@ -23,7 +26,6 @@ class ProcessingJobRecord(Base):
     user_id: Mapped[str] = mapped_column(String, index=True)
     platform_job_id: Mapped[str] = mapped_column(String, index=True)
     parameters: Mapped[str] = mapped_column(String, index=False)
-    result_link: Mapped[Optional[str]] = mapped_column(String, index=False)
     service: Mapped[str] = mapped_column(String, index=True)
     created: Mapped[datetime.datetime] = mapped_column(
         DateTime, default=datetime.datetime.utcnow, index=True
@@ -116,14 +118,12 @@ def update_job_status_by_id(
         )
 
 
-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}"
-    )
+def update_job_result_by_id(database: Session, job_id: int, result: Collection):
+    logger.info(f"Updating the result link of processing job with ID {job_id}")
     job = get_job_by_id(database, job_id)
 
     if job:
-        job.result_link = result_link
+        job.result = json.dumps(result)
         database.commit()
         database.refresh(job)
     else:

app/platforms/base.py

@@ -3,6 +3,8 @@
 from app.schemas.enum import ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
 
+from stac_pydantic import Collection
+
 
 class BaseProcessingPlatform(ABC):
     """
@@ -36,12 +38,12 @@ def get_job_status(
         pass
 
     @abstractmethod
-    def get_job_result_url(self, job_id: str, details: ServiceDetails) -> str:
+    def get_job_results(self, job_id: str, details: ServiceDetails) -> Collection:
         """
         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
+        :return: STAC collection representing the results.
         """
         pass

app/platforms/implementations/ogc_api_process.py

@@ -2,6 +2,7 @@
 from app.platforms.dispatcher import register_platform
 from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
+from stac_pydantic import Collection
 
 
 @register_platform(ProcessTypeEnum.OGC_API_PROCESS)
@@ -21,7 +22,7 @@ def get_job_status(
             "OGC API Process job status retrieval not implemented yet."
         )
 
-    def get_job_result_url(self, job_id: str, details: ServiceDetails) -> str:
+    def get_job_results(self, job_id: str, details: ServiceDetails) -> Collection:
         raise NotImplementedError(
             "OGC API Process job result retrieval not implemented yet."
         )

app/platforms/implementations/openeo.py

@@ -13,6 +13,7 @@
 from app.platforms.dispatcher import register_platform
 from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
+from stac_pydantic import Collection
 
 load_dotenv()
 
@@ -201,12 +202,12 @@ def get_job_status(
                 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:
+    def get_job_results(self, job_id: str, details: ServiceDetails) -> Collection:
         try:
             logger.debug(f"Fetching job result for openEO job with ID {job_id}")
             connection = self._setup_connection(details.endpoint)
             job = connection.job(job_id)
-            return job.get_results_metadata_url(full=True)
+            return Collection(**job.get_results().get_metadata())
         except Exception as e:
             logger.exception(
                 f"Failed to fetch result url for for openEO job with ID {job_id}"

app/routers/unit_jobs.py

@@ -11,7 +11,13 @@
     ProcessingJobSummary,
     ServiceDetails,
 )
-from app.services.processing import create_processing_job, get_processing_job_by_user_id
+from app.services.processing import (
+    create_processing_job,
+    get_processing_job_by_user_id,
+    get_processing_job_results,
+)
+
+from stac_pydantic import Collection
 
 # from app.auth import get_current_user
 
@@ -121,3 +127,32 @@ async def get_job(
             detail=f"Processing job {job_id} not found",
         )
     return job
+
+
+@router.get(
+    "/unit_jobs/{job_id}/results",
+    tags=["Unit Jobs"],
+    responses={404: {"description": "Processing job not found"}},
+)
+async def get_job_results(
+    job_id: int, db: Session = Depends(get_db), user: str = "foobar"
+) -> Collection | None:
+    try:
+        result = get_processing_job_results(db, job_id, user)
+        if not result:
+            logger.error(
+                f"Result for processing job {job_id} not found for user {user}"
+            )
+            raise HTTPException(
+                status_code=404,
+                detail=f"Result for processing job {job_id} not found",
+            )
+        return result
+    except HTTPException as e:
+        raise e
+    except Exception as e:
+        logger.exception(f"Error creating processing job for user {user}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"An error occurred while creating the processing job: {e}",
+        )

app/schemas/unit_job.py

@@ -1,5 +1,4 @@
 from datetime import datetime
-from typing import Optional
 
 from pydantic import BaseModel, Field
 
@@ -48,14 +47,6 @@ class ProcessingJobSummary(BaseModel):
         description="JSON representing the parameters for the service execution",
         examples=[{"param1": "value1", "param2": "value2"}],
     )
-    result_link: Optional[str] = Field(
-        ...,
-        description="URL to the results of the processing job",
-        examples=[
-            "https://openeofed.dataspace.copernicus.eu/jobs/"
-            "cdse-j-25082106161041f1a151bd539f614130/results"
-        ],
-    )
 
 
 class ProcessingJobDetails(BaseModel):

app/schemas/upscale_task.py

@@ -44,15 +44,13 @@ class UpscalingTaskDetails(BaseModel):
                     label=ProcessTypeEnum.OPENEO,
                     status=ProcessingStatusEnum.FINISHED,
                     parameters={"param1": "value1", "param2": "value2"},
-                    result_link="https://foo.bar",
                 ),
                 ProcessingJobSummary(
                     id=1,
                     title="Upscaling Job 2",
                     label=ProcessTypeEnum.OPENEO,
                     status=ProcessingStatusEnum.RUNNING,
                     parameters={"param1": "value1", "param2": "value2"},
-                    result_link=None,
                 ),
             ]
         ],

app/services/processing.py

@@ -7,7 +7,6 @@
     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
@@ -21,6 +20,7 @@
     ServiceDetails,
 )
 
+from stac_pydantic import Collection
 
 INACTIVE_JOB_STATUSES = {
     ProcessingStatusEnum.CANCELED,
@@ -50,7 +50,6 @@ def create_processing_job(
         user_id=user,
         platform_job_id=job_id,
         parameters=json.dumps(request.parameters),
-        result_link=None,
         service=request.service.model_dump_json(),
         upscaling_task_id=upscaling_task_id,
     )
@@ -61,7 +60,6 @@ def create_processing_job(
         label=request.label,
         status=record.status,
         parameters=request.parameters,
-        result_link=None,
     )
 
 
@@ -74,11 +72,17 @@ 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)
-    return platform.get_job_result_url(job.platform_job_id, details)
+def get_processing_job_results(
+    database: Session, job_id: int, user_id: str
+) -> Collection | None:
+    record = get_job_by_user_id(database, job_id, user_id)
+    if not record:
+        return None
+
+    logger.info(f"Retrieving job result for job: {record.platform_job_id}")
+    platform = get_processing_platform(record.label)
+    details = ServiceDetails.model_validate_json(record.service)
+    return platform.get_job_results(record.platform_job_id, details)
 
 
 def _refresh_job_status(
@@ -105,19 +109,13 @@ def get_processing_jobs_by_user_id(
         if record.status not in INACTIVE_JOB_STATUSES:
             record = _refresh_job_status(database, record)
 
-        # Update the result if the job is finished and results weren't retrieved yet
-        if record.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=record.status,
                 parameters=json.loads(record.parameters),
-                result_link=record.result_link,
             )
         )
     return jobs
@@ -141,7 +139,6 @@ def get_processing_job_by_user_id(
         status=record.status,
         service=ServiceDetails.model_validate_json(record.service or "{}"),
         parameters=json.loads(record.parameters or "{}"),
-        result_link=record.result_link,
         created=record.created,
         updated=record.updated,
     )