feat: added dynamic registration of platforms

Author: JanssenBrm

app/main.py

@@ -1,10 +1,14 @@
 from fastapi import FastAPI
+
+from app.platforms.dispatcher import load_processing_platforms
 from .config.logger import setup_logging
 from .config.settings import settings
 from .routers import jobs_status, unit_jobs, health
 
 setup_logging()
 
+load_processing_platforms()
+
 app = FastAPI(
     title=settings.app_name,
     description=settings.app_description,

app/platforms/dispatcher.py

@@ -1,9 +1,29 @@
 
+import importlib
+import logging
+import app.platforms.implementations 
+import pkgutil
+from typing import Dict, Type
 from app.platforms.base import BaseProcessingPlatform
-from app.platforms.ogc_api_process import OGCAPIProcessPlatform
-from app.platforms.openeo import OpenEOPlatform 
 from app.schemas import ProcessType
 
+PROCESSING_PLATFORMS: Dict[ProcessType, Type[BaseProcessingPlatform]] = {}
+
+logger = logging.getLogger(__name__)
+
+def load_processing_platforms():
+    """Dynamically load all processing platform implementations."""
+    for _, module_name, _ in pkgutil.iter_modules(app.platforms.implementations.__path__):
+        importlib.import_module(f"app.platforms.implementations.{module_name}")
+
+def register_processing_platform(service_type: ProcessType, cls: Type[BaseProcessingPlatform]):
+    """"Register a new processing platform class for a specific service type.
+    
+    :param service_type: The type of service for which to register the platform.
+    :param cls: The class that implements BaseProcessingPlatform.
+    """
+    logger.debug(f"Registering processing platform with class {cls} for service type: {service_type}")
+    PROCESSING_PLATFORMS[service_type] = cls
 
 def get_processing_platform(service_type: ProcessType) -> BaseProcessingPlatform:
     """
@@ -12,9 +32,7 @@ def get_processing_platform(service_type: ProcessType) -> BaseProcessingPlatform
     :param service_type: The type of service for which to get the processing platform.
     :return: An instance of a class that implements BaseProcessingPlatform.
     """
-    if service_type == ProcessType.OPENEO:
-        return OpenEOPlatform()
-    elif service_type == ProcessType.OGC_API_PROCESS:
-        return OGCAPIProcessPlatform()
-    else:
+    try:
+        return PROCESSING_PLATFORMS[service_type]()
+    except KeyError:
         raise ValueError(f"Unsupported service type: {service_type}")

app/platforms/ogc_api_process.py …forms/implementations/ogc_api_process.pyapp/platforms/ogc_api_process.py renamed to app/platforms/implementations/ogc_api_process.py app/platforms/ogc_api_process.py renamed to app/platforms/implementations/ogc_api_process.py

@@ -2,7 +2,8 @@
 import logging
 
 from app.platforms.base import BaseProcessingPlatform
-from app.schemas import ProcessingJobSummary, ServiceDetails
+from app.platforms.dispatcher import register_processing_platform
+from app.schemas import ProcessType, ProcessingJobSummary, ServiceDetails
 
 
 logger = logging.getLogger(__name__)
@@ -22,4 +23,6 @@ def execute_job(self, title: str, details: ServiceDetails, parameters: dict) ->
         :param parameters: The parameters required for the job execution.
         :return: A ProcessingJobSummary object containing the job details.
         """
-        raise NotImplementedError("OGC API Process job execution not implemented yet.")
+        raise NotImplementedError("OGC API Process job execution not implemented yet.")
+        
+register_processing_platform(ProcessType.OGC_API_PROCESS, OGCAPIProcessPlatform)

app/platforms/openeo.py app/platforms/implementations/openeo.pyapp/platforms/openeo.py renamed to app/platforms/implementations/openeo.py

@@ -1,22 +1,32 @@
 
 import logging
 import os
-import openeo
 import re
-
 import urllib
 
+import openeo
 import requests
+from dotenv import load_dotenv
 
 from app.platforms.base import BaseProcessingPlatform
-from app.schemas import ProcessingJobSummary, ProcessingStatusEnum, ServiceDetails
-
-from dotenv import load_dotenv
+from app.platforms.dispatcher import register_processing_platform
+from app.schemas import (
+    ProcessingJobSummary,
+    ProcessingStatusEnum,
+    ProcessType,
+    ServiceDetails,
+)
 
 load_dotenv()
-
 logger = logging.getLogger(__name__)
 
+# Constants
+BACKEND_AUTH_ENV_MAP = {
+    "openeo.dataspace.copernicus.eu": "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED",
+    "openeofed.dataspace.copernicus.eu": "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED",
+}
+
+
 class OpenEOPlatform(BaseProcessingPlatform):
     """
     OpenEO processing platform implementation.
@@ -32,8 +42,9 @@ def _setup_connection(self, url: str) -> None:
         connection = openeo.connect(url)
         provider_id, client_id, client_secret = self._get_client_credentials(url)
 
-        # connection.authenticate_oidc_device()
-                                    
+        # @TODO: Remove the line below as this is only for local testing
+        # connection.authenticate_oidc_device()  
+                                     
         connection.authenticate_oidc_client_credentials(
                 provider_id=provider_id,
                 client_id=client_id,
@@ -49,29 +60,31 @@ def _get_client_credentials(self, url: str) -> tuple[str, str, str]:
         :param url: The URL of the OpenEO backend.
         :return: A tuple containing provider ID, client ID, and client secret.
         """
-        auth_env_var = self._get_client_credentials_env_var(url)
-        if auth_env_var not in os.environ:
-            raise ValueError(f"Environment variable {auth_env_var} not set.")
+        env_var = self._get_client_credentials_env_var(url)
+        credentials_str = os.getenv(env_var)
+
+        if not credentials_str:
+            raise ValueError(f"Environment variable {env_var} not set.")
 
-        client_credentials = os.environ[auth_env_var]
-        return client_credentials.split("/", 2)
+        parts = credentials_str.split("/", 2)
+        if len(parts) != 3:
+            raise ValueError(
+                f"Invalid client credentials format in {env_var}, expected 'provider_id/client_id/client_secret'."
+            )
+        return tuple(parts)
 
     def _get_client_credentials_env_var(self, url: str) -> str:
         """
         Get client credentials env var name for a given backend URL.
         """
         if not re.match(r"https?://", url):
             url = f"https://{url}"
-        parsed = urllib.parse.urlparse(url)
 
-        hostname = parsed.hostname
-        if hostname in {
-            "openeo.dataspace.copernicus.eu",
-            "openeofed.dataspace.copernicus.eu",
-        }:
-            return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED"
-        else:
-            raise ValueError(f"Unsupported backend: {url=} ({hostname=})")
+        hostname = urllib.parse.urlparse(url).hostname
+        if not hostname or hostname not in BACKEND_AUTH_ENV_MAP:
+            raise ValueError(f"Unsupported backend: {url} (hostname={hostname})")
+
+        return BACKEND_AUTH_ENV_MAP[hostname]
 
     def _get_process_id(self, url: str) -> str:
         """        
@@ -81,8 +94,18 @@ def _get_process_id(self, url: str) -> str:
         :return: The process ID extracted from the JSON file.
         """
         logger.debug(f"Fetching process ID from {url}")
-        response = requests.get(url).json()
-        return response.get("id")
+        try:
+            response = requests.get(url)
+            response.raise_for_status()
+        except requests.RequestException as e:
+            logger.error(f"Error fetching process ID from {url}: {e}")
+            raise ValueError(f"Failed to fetch process ID from {url}")
+
+        process_id = response.json().get("id")
+        if not process_id:
+            raise ValueError(f"No 'id' field found in process definition at {url}")
+
+        return process_id
 
     def execute_job(self, title: str, details: ServiceDetails , parameters: dict) -> ProcessingJobSummary:
         """
@@ -96,10 +119,12 @@ def execute_job(self, title: str, details: ServiceDetails , parameters: dict) ->
 
         try:
             process_id = self._get_process_id(details.application)
-            if not process_id:
-                raise ValueError(f"Process ID not found for service: {details.service}")
 
-            logger.debug(f"Executing OpenEO job with title={title}, service={details}, process_id={process_id} and parameters={parameters}")
+            logger.debug(
+                f"Executing OpenEO job with title={title}, service={details}, "
+                f"process_id={process_id}, parameters={parameters}"
+            )
+            
             connection = self._setup_connection(details.service)
             service = connection.datacube_from_process(
                 process_id=process_id,
@@ -116,4 +141,7 @@ def execute_job(self, title: str, details: ServiceDetails , parameters: dict) ->
             )
         except Exception as e:
             logger.exception(f"Failed to execute openEO job: {e}")  
-            raise Exception("Failed to execute openEO job")
+            raise SystemError("Failed to execute openEO job")
+
+
+register_processing_platform(ProcessType.OPENEO, OpenEOPlatform)