refactor: used decorator for platform registration

Author: JanssenBrm

app/platforms/dispatcher.py

@@ -11,6 +11,17 @@
 logger = logging.getLogger(__name__)
 
 
+def register_platform(service_type: ProcessTypeEnum):
+    def decorator(cls: Type[BaseProcessingPlatform]):
+        logger.debug(
+            f"Registering processing platform with class {cls} for service type: {service_type}"
+        )
+        PROCESSING_PLATFORMS[service_type] = cls
+        return cls
+
+    return decorator
+
+
 def load_processing_platforms():
     """Dynamically load all processing platform implementations."""
     for _, module_name, _ in pkgutil.iter_modules(
@@ -19,20 +30,6 @@ def load_processing_platforms():
         importlib.import_module(f"app.platforms.implementations.{module_name}")
 
 
-def register_processing_platform(
-    service_type: ProcessTypeEnum, 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: ProcessTypeEnum) -> BaseProcessingPlatform:
     """
     Factory function to get the appropriate processing platform based on the service type.

app/platforms/implementations/ogc_api_process.py

@@ -1,13 +1,14 @@
 import logging
 
 from app.platforms.base import BaseProcessingPlatform
-from app.platforms.dispatcher import register_processing_platform
+from app.platforms.dispatcher import register_platform
 from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
 
 logger = logging.getLogger(__name__)
 
 
+@register_platform(ProcessTypeEnum.OGC_API_PROCESS)
 class OGCAPIProcessPlatform(BaseProcessingPlatform):
     """
     OGC API Process processing platform implementation.
@@ -23,6 +24,3 @@ def get_job_status(
         raise NotImplementedError(
             "OGC API Process job status retrieval not implemented yet."
         )
-
-
-register_processing_platform(ProcessTypeEnum.OGC_API_PROCESS, OGCAPIProcessPlatform)

app/platforms/implementations/openeo.py

@@ -8,7 +8,7 @@
 from dotenv import load_dotenv
 
 from app.platforms.base import BaseProcessingPlatform
-from app.platforms.dispatcher import register_processing_platform
+from app.platforms.dispatcher import register_platform 
 from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
 
@@ -22,6 +22,7 @@
 }
 
 
+@register_platform(ProcessTypeEnum.OPENEO)
 class OpenEOPlatform(BaseProcessingPlatform):
     """
     OpenEO processing platform implementation.
@@ -160,6 +161,3 @@ def get_job_status(
             raise SystemError(
                 f"Failed to fetch status openEO job with ID {job_id}"
             ) from e
-
-
-register_processing_platform(ProcessTypeEnum.OPENEO, OpenEOPlatform)

tests/platforms/test_dispatcher.py

@@ -27,7 +27,7 @@ def clear_registry():
 
 
 def test_register_processing_platform():
-    dispatcher.register_processing_platform(ProcessTypeEnum.OPENEO, DummyPlatform)
+    dispatcher.register_platform(ProcessTypeEnum.OPENEO)(DummyPlatform)
     assert dispatcher.PROCESSING_PLATFORMS[ProcessTypeEnum.OPENEO] is DummyPlatform