refactor: service details refactor to use endpoint iso service

Author: JanssenBrm

alembic/versions/6105be2a2b21_processing_jobs_rename_of_service_.py

@@ -0,0 +1,38 @@
+"""processing_jobs: rename of service_record field
+
+Revision ID: 6105be2a2b21
+Revises: a0ca6614c7c1
+Create Date: 2025-08-20 16:55:05.666856
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '6105be2a2b21'
+down_revision: Union[str, Sequence[str], None] = 'a0ca6614c7c1'
+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.add_column('processing_jobs', sa.Column('service', sa.String(), nullable=False))
+    op.drop_index(op.f('ix_processing_jobs_service_record'), table_name='processing_jobs')
+    op.create_index(op.f('ix_processing_jobs_service'), 'processing_jobs', ['service'], unique=False)
+    op.drop_column('processing_jobs', 'service_record')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('processing_jobs', sa.Column('service_record', sa.VARCHAR(), autoincrement=False, nullable=False))
+    op.drop_index(op.f('ix_processing_jobs_service'), table_name='processing_jobs')
+    op.create_index(op.f('ix_processing_jobs_service_record'), 'processing_jobs', ['service_record'], unique=False)
+    op.drop_column('processing_jobs', 'service')
+    # ### end Alembic commands ###

app/database/models/processing_job.py

@@ -26,7 +26,7 @@ class ProcessingJobRecord(Base):
     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_record: Mapped[str] = mapped_column(String, index=True)
+    service: Mapped[str] = mapped_column(String, index=True)
     created: Mapped[datetime.datetime] = mapped_column(
         DateTime, default=datetime.datetime.utcnow, index=True
     )

app/platforms/implementations/openeo.py

@@ -111,7 +111,7 @@ def execute_job(self, title: str, details: ServiceDetails, parameters: dict) ->
                 f"process_id={process_id}, parameters={parameters}"
             )
 
-            connection = self._setup_connection(details.service)
+            connection = self._setup_connection(details.endpoint)
             service = connection.datacube_from_process(
                 process_id=process_id, namespace=details.application, **parameters
             )
@@ -153,7 +153,7 @@ def get_job_status(
     ) -> ProcessingStatusEnum:
         try:
             logger.debug(f"Fetching job status for openEO job with ID {job_id}")
-            connection = self._setup_connection(details.service)
+            connection = self._setup_connection(details.endpoint)
             job = connection.job(job_id)
             return self._map_openeo_status(job.status())
         except Exception as e:
@@ -165,9 +165,9 @@ def get_job_status(
     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)
+            connection = self._setup_connection(details.endpoint)
             job = connection.job(job_id)
-            return f"{details.service}{job.get_results_metadata_url()}"
+            return f"{details.endpoint}{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}"

app/schemas/unit_job.py

@@ -7,7 +7,7 @@
 
 
 class ServiceDetails(BaseModel):
-    service: str
+    endpoint: str
     application: str
 
 

app/services/processing.py

@@ -43,7 +43,7 @@ def create_processing_job(
         platform_job_id=job_id,
         parameters=json.dumps(summary.parameters),
         result_link=None,
-        service_record=summary.service.model_dump_json(),
+        service=summary.service.model_dump_json(),
     )
     record = save_job_to_db(database, record)
     return ProcessingJobSummary(
@@ -54,14 +54,14 @@ def create_processing_job(
 def get_job_status(job: ProcessingJobRecord) -> ProcessingStatusEnum:
     logger.info(f"Retrieving job status for job: {job.platform_job_id}")
     platform = get_processing_platform(job.label)
-    details = ServiceDetails.model_validate_json(job.service_record)
+    details = ServiceDetails.model_validate_json(job.service)
     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)
+    details = ServiceDetails.model_validate_json(job.service)
     return platform.get_job_result_url(job.platform_job_id, details)
 
 
@@ -115,7 +115,7 @@ def get_processing_job_by_user_id(
         title=record.title,
         label=record.label,
         status=record.status,
-        service=ServiceDetails.model_validate_json(record.service_record or "{}"),
+        service=ServiceDetails.model_validate_json(record.service or "{}"),
         parameters=json.loads(record.parameters or "{}"),
         result_link=record.result_link,
         created=record.created,

tests/conftest.py

@@ -32,7 +32,7 @@ def fake_processing_job_request():
     return BaseJobRequest(
         title="Test Job",
         label=ProcessTypeEnum.OPENEO,
-        service=ServiceDetails(service="foo", application="bar"),
+        service=ServiceDetails(endpoint="foo", application="bar"),
         parameters={},
     )
 
@@ -66,7 +66,7 @@ def fake_processing_job_record(
     record = ProcessingJobRecord(
         **(fake_processing_job_summary.model_dump()),
         platform_job_id="platform-job-1",
-        service_record='{"service":"foo","application":"bar"}',
+        service='{"endpoint":"foo","application":"bar"}',
         parameters="{}",
         result_link="https://foo.bar",
         created=datetime.now(),

tests/platforms/test_openeo_platform.py

@@ -33,8 +33,7 @@ def mock_env(monkeypatch):
 @pytest.fixture
 def service_details():
     return ServiceDetails(
-        service="https://openeo.dataspace.copernicus.eu",
-        service_id="service-1",
+        endpoint="https://openeo.dataspace.copernicus.eu",
         application="https://example.com/process.json",
     )
 
@@ -107,7 +106,7 @@ def test_execute_job_success(mock_pid, mock_connect, platform, service_details):
     )
 
     assert job_id == "job123"
-    mock_connect.assert_called_once_with(service_details.service)
+    mock_connect.assert_called_once_with(service_details.endpoint)
 
 
 @patch("app.platforms.implementations.openeo.openeo.connect")
@@ -145,7 +144,7 @@ def test_map_openeo_status(openeo_status, expected_enum):
 def test_get_job_status_success(mock_connection, platform):
     mock_connection.return_value = DummyOpenEOClient()
 
-    details = ServiceDetails(service="foo", application="bar")
+    details = ServiceDetails(endpoint="foo", application="bar")
     result = platform.get_job_status("job123", details)
 
     assert result == ProcessingStatusEnum.RUNNING
@@ -155,7 +154,7 @@ def test_get_job_status_success(mock_connection, platform):
 def test_get_job_status_error(mock_connection, platform):
     mock_connection.side_effect = RuntimeError("Connection error")
 
-    details = ServiceDetails(service="foo", application="bar")
+    details = ServiceDetails(endpoint="foo", application="bar")
     with pytest.raises(SystemError) as exc_info:
         platform.get_job_status("job123", details)
 
@@ -166,7 +165,7 @@ def test_get_job_status_error(mock_connection, platform):
 def test_get_job_result_success(mock_connection, platform):
     mock_connection.return_value = DummyOpenEOClient()
 
-    details = ServiceDetails(service="foo", application="bar")
+    details = ServiceDetails(endpoint="foo", application="bar")
     result = platform.get_job_result_url("job123", details)
 
     assert result == "foo/job/results"
@@ -176,7 +175,7 @@ def test_get_job_result_success(mock_connection, platform):
 def test_get_job_url_error(mock_connection, platform):
     mock_connection.side_effect = RuntimeError("Connection error")
 
-    details = ServiceDetails(service="foo", application="bar")
+    details = ServiceDetails(endpoint="foo", application="bar")
     with pytest.raises(SystemError) as exc_info:
         platform.get_job_result_url("job123", details)
 

tests/services/test_processing.py

@@ -26,7 +26,7 @@ def make_job_request():
         label=ProcessTypeEnum.OPENEO,
         title="Test Job",
         service=ServiceDetails(
-            service="dummy-service-id", application="dummy-application"
+            endpoint="dummy-service-id", application="dummy-application"
         ),
         parameters={"param": 1},
     )
@@ -95,7 +95,7 @@ def test_get_processing_jobs_with_active_and_inactive_statuses(
         label=ProcessTypeEnum.OGC_API_PROCESS,
         title="Finished Job",
         status=ProcessingStatusEnum.FAILED,
-        service_record=json.dumps({"foo": "bar"}),
+        service=json.dumps({"foo": "bar"}),
     )
     mock_get_jobs.return_value = [fake_processing_job_record, inactive_job]
     mock_get_job_status.return_value = ProcessingStatusEnum.RUNNING
@@ -149,15 +149,15 @@ def test_get_processing_jobs_with_finished_statuses(
         label=ProcessTypeEnum.OGC_API_PROCESS,
         title="Finished Job",
         status=ProcessingStatusEnum.FINISHED,
-        service_record=json.dumps({"foo": "bar"}),
+        service=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"}),
+        service=json.dumps({"foo": "bar"}),
         result_link="https://foo.bar",
     )
     mock_get_jobs.return_value = [finished_job_no_result, finished_job_result]
@@ -203,7 +203,7 @@ def test_get_job_result_from_platform(mock_get_platform, fake_processing_job_rec
 def test_get_processing_job_by_user_id(mock_get_job, fake_db_session):
 
     fake_service_details = {
-        "service": "https://openeofed.dataspace.copernicus.eu",
+        "endpoint": "https://openeofed.dataspace.copernicus.eu",
         "application": "https://raw.githubusercontent.com/ESA-APEx/apex_algorithms/"
         "32ea3c9a6fa24fe063cb59164cd318cceb7209b0/openeo_udp/variabilitymap/"
         "variabilitymap.json",
@@ -219,7 +219,7 @@ def test_get_processing_job_by_user_id(mock_get_job, fake_db_session):
         result_link=None,
         created="2025-08-11T10:00:00",
         updated="2025-08-11T10:00:00",
-        service_record=json.dumps(fake_service_details),
+        service=json.dumps(fake_service_details),
     )
     mock_get_job.return_value = fake_result
 
@@ -231,7 +231,7 @@ def test_get_processing_job_by_user_id(mock_get_job, fake_db_session):
     assert result.title == "Test Job"
     assert result.status == ProcessingStatusEnum.CREATED
     assert isinstance(result.service, ServiceDetails)
-    assert result.service.service == fake_service_details["service"]
+    assert result.service.endpoint == fake_service_details["endpoint"]
     assert result.service.application == fake_service_details["application"]
     assert result.parameters == {"param1": "value1"}