chore: initial setup of api framework

JanssenBrm · JanssenBrm · commit 1b0594c6e6da · 2025-08-08T16:25:06.000+02:00
diff --git a/README.md b/README.md
@@ -1,2 +1,22 @@
-# APEx Dispatch API
+# APEx Dispatch API (FastAPI)
 Implementation of the APEx Upscaling Service API
+
+
+## Running the API locally
+
+1. Install the required dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+2. Set up your environment variables in a `.env.local` file
+3. Run the FastAPI application:
+   ```bash
+   uvicorn app.main:app --reload
+   ```
+
+## Running Tests
+
+To run the tests, use the following command:
+```bash
+pytest
+```
diff --git a/app/__init__.py b/app/__init__.py
diff --git a/app/auth.py b/app/auth.py
@@ -0,0 +1,17 @@
+from fastapi_keycloak import FastAPIKeycloak
+from .config.settings import settings
+
+# create the FastAPIKeycloak instance — used to protect routes
+# The server_url must include trailing slash for library
+keycloak = FastAPIKeycloak(
+    server_url=str(settings.keycloak_server_url),
+    client_id=settings.keycloak_client_id,
+    client_secret=settings.keycloak_client_secret,
+    admin_client_secret=settings.keycloak_client_secret,  # optional for admin operations
+    realm=settings.keycloak_realm,
+    callback_uri="http://localhost:8000/callback",  # for auth code flow if needed
+)
+
+# expose a helper dependency for current user
+get_current_user = keycloak.get_current_user
+get_current_active_user = keycloak.get_current_user
diff --git a/app/config/logger.py b/app/config/logger.py
@@ -0,0 +1,34 @@
+import logging
+import logging.config
+
+
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,  # keeps Uvicorn's loggers
+    "formatters": {
+        "default": {
+            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        },
+    },
+    "handlers": {
+        "console": {
+            "class": "logging.StreamHandler",
+            "formatter": "default",
+        },
+    },
+    "root": {  # applies to all loggers unless overridden
+        "level": "INFO",
+        "handlers": ["console"],
+    },
+    "loggers": {
+        "uvicorn": {"level": "INFO"},
+        "uvicorn.error": {"level": "INFO"},
+        "uvicorn.access": {"level": "INFO"},
+        # custom API loggers
+        "routers": {"level": "DEBUG"},      # all your routers
+        "services": {"level": "DEBUG"},     # all your services
+    },
+}
+
+def setup_logging():
+    logging.config.dictConfig(LOGGING_CONFIG)
diff --git a/app/config/settings.py b/app/config/settings.py
@@ -0,0 +1,23 @@
+from pydantic import AnyHttpUrl, ConfigDict, Field
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    app_name: str = Field("", json_schema_extra={"env": "APP_NAME"})
+    app_description: str = Field("", json_schema_extra={"env":"APP_DESCRIPTION"})
+    env: str = Field("development", json_schema_extra={"env":"APP_ENV"})
+
+    # Keycloak / OIDC
+    keycloak_server_url: AnyHttpUrl = Field(None, json_schema_extra={"env":"KEYCLOAK_SERVER_URL"})
+    keycloak_realm: str = Field(None, json_schema_extra={"env":"KEYCLOAK_REALM"})
+    keycloak_client_id: str = Field(None, json_schema_extra={"env":"KEYCLOAK_CLIENT_ID"})
+    keycloak_client_secret: str | None = Field(None, json_schema_extra={"env":"KEYCLOAK_CLIENT_SECRET"})
+
+
+    model_config = ConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="allow",
+    )
+
+
+settings = Settings()
diff --git a/app/main.py b/app/main.py
@@ -0,0 +1,20 @@
+from fastapi import FastAPI
+from .config.logger import setup_logging
+from .config.settings import settings
+from .routers import jobs_status, unit_jobs, health
+
+setup_logging()
+
+app = FastAPI(
+    title=settings.app_name,
+    description=settings.app_description,
+    version="1.0.0", 
+)
+
+# Register Keycloak - must be done after FastAPI app creation
+# keycloak.register(app, prefix="/auth")  # mounts OIDC endpoints for login if needed
+
+# include routers
+app.include_router(jobs_status.router)
+app.include_router(unit_jobs.router)
+app.include_router(health.router)
diff --git a/app/platforms/base.py b/app/platforms/base.py
@@ -0,0 +1,21 @@
+from abc import ABC, abstractmethod
+
+from app.schemas import ProcessingJobSummary
+
+
+class BaseProcessingPlatform(ABC):
+    """
+    Abstract base class for processing platforms.
+    Defines the interface for processing jobs and managing platform-specific configurations.
+    """
+
+    @abstractmethod
+    def execute_job(self, service_id: str, parameters: dict) -> ProcessingJobSummary:
+        """
+        Execute a processing job on the platform with the given service ID and parameters.
+
+        :param service_id: The ID of the service to execute.
+        :param parameters: The parameters required for the job execution.
+        :return: A ProcessingJobSummary object containing the job details.
+        """
+        pass
diff --git a/app/platforms/dispatcher.py b/app/platforms/dispatcher.py
@@ -0,0 +1,20 @@
+
+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
+
+
+def get_processing_platform(service_type: ProcessType) -> BaseProcessingPlatform:
+    """
+    Factory function to get the appropriate processing platform based on the service type.
+    
+    :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:
+        raise ValueError(f"Unsupported service type: {service_type}")
diff --git a/app/platforms/ogc_api_process.py b/app/platforms/ogc_api_process.py
@@ -0,0 +1,25 @@
+
+import logging
+
+from app.platforms.base import BaseProcessingPlatform
+from app.schemas import ProcessingJobSummary
+
+
+logger = logging.getLogger(__name__)
+
+class OGCAPIProcessPlatform(BaseProcessingPlatform):
+    """
+    OGC API Process processing platform implementation.
+    This class handles the execution of processing jobs on the OGC API Process platform.
+    """
+
+    def execute_job(self, service_id: str, parameters: dict) -> ProcessingJobSummary:
+        """
+        Execute a processing job on the OGC API Process platform with the given service ID and parameters.
+
+        :param service_id: The ID of the service to execute.
+        :param parameters: The parameters required for the job execution.
+        :return: A ProcessingJobSummary object containing the job details.
+        """
+        logger.debug(f"Executing OGC API Process job with service_id={service_id} and parameters={parameters}")
+        raise NotImplementedError("OGC API Process job execution not implemented yet.")
diff --git a/app/platforms/openeo.py b/app/platforms/openeo.py
@@ -0,0 +1,25 @@
+
+import logging
+
+from app.platforms.base import BaseProcessingPlatform
+from app.schemas import ProcessingJobSummary
+
+
+logger = logging.getLogger(__name__)
+
+class OpenEOPlatform(BaseProcessingPlatform):
+    """
+    OpenEO processing platform implementation.
+    This class handles the execution of processing jobs on the OpenEO platform.
+    """
+
+    def execute_job(self, service_id: str, parameters: dict) -> ProcessingJobSummary:
+        """
+        Execute a processing job on the OpenEO platform with the given service ID and parameters.
+
+        :param service_id: The ID of the service to execute.
+        :param parameters: The parameters required for the job execution.
+        :return: A ProcessingJobSummary object containing the job details.
+        """
+        logger.debug(f"Executing OpenEO job with service_id={service_id} and parameters={parameters}")
+        raise NotImplementedError("OpenEO job execution not implemented yet.")
diff --git a/app/routers/health.py b/app/routers/health.py
@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+router = APIRouter()
+
+@router.get("/health")
+async def health():
+    return {"status": "ok"}
diff --git a/app/routers/jobs_status.py b/app/routers/jobs_status.py
@@ -0,0 +1,21 @@
+import logging
+from fastapi import APIRouter, Query
+from app.schemas import JobsStatusResponse
+
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+@router.get("/jobs_status", response_model=JobsStatusResponse, tags=["Upscale Tasks", "Unit Jobs"], summary="Get a list of all upscaling tasks & processing jobs for the authenticated user")
+async def jobs_status(
+    limit: int = Query(50, ge=1),
+    offset: int = Query(0, ge=0),
+):
+    """
+    Return combined list of upscaling tasks and processing jobs for the authenticated user.
+    """
+    logger.debug(f"Fetching jobs status with limit={limit}, offset={offset}")
+    return JobsStatusResponse(
+        upscalingTasks=[],
+        processingJobs=[]
+    )
diff --git a/app/routers/unit_jobs.py b/app/routers/unit_jobs.py
@@ -0,0 +1,23 @@
+import logging
+from app.services.processing import create_processing_job
+from fastapi import APIRouter, status, HTTPException
+from app.schemas import BaseJobRequest, ProcessingJobSummary 
+# from app.auth import get_current_user
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+
+@router.post("/unit_jobs", response_model=ProcessingJobSummary, status_code=status.HTTP_201_CREATED, tags=["Unit Jobs"], summary="Create a new processing job")
+async def create_unit_job(
+    payload: BaseJobRequest, 
+    ): 
+    """Create a new processing job with the provided data."""
+    try:
+        return create_processing_job(payload)
+    except Exception as e:
+        logger.error(f"Error creating unit job: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"An error occurred while creating the processing job: {e}"
+        )
diff --git a/app/schemas.py b/app/schemas.py
@@ -0,0 +1,92 @@
+from enum import Enum
+from typing import List
+from pydantic import BaseModel
+
+class ProcessType(str, Enum):
+    OPENEO = "openeo"
+    OGC_API_PROCESS = "ogc_api_process"
+    
+class ProcessingStatusEnum(str, Enum):
+    CREATED = "created"
+    RUNNING = "running"
+    FINISHED = "finished"
+    FAILED = "failed"
+
+
+# # Tile / GeoJSON placeholders
+# class Tile(BaseModel):
+#     type: Optional[str] = "Feature"
+#     geometry: Optional[dict] = {}
+
+
+# class TileRequest(BaseModel):
+#     aoi: dict
+#     grid: str
+
+
+# class TileResponse(BaseModel):
+#     tiles: List[Tile] = []
+
+
+# Service / parameters
+class ServiceDetails(BaseModel):
+    id: str
+
+
+
+# # Process type, status
+# class ProcessType(str):
+#     pass
+
+
+# class ProcessingStatus(str):
+#     pass
+
+
+class ProcessingJobSummary(BaseModel):
+    id: int
+    title: str
+    status: ProcessingStatusEnum
+
+
+# class ProcessingJobDetails(BaseModel):
+#     label: str
+#     service: ServiceDetails
+#     parameters: ServiceParameters
+
+
+# class ProcessingJob(ProcessingJobSummary, ProcessingJobDetails):
+#     pass
+
+
+class UpscalingTaskSummary(BaseModel):
+    id: int
+    title: str
+    status: ProcessingStatusEnum
+
+
+# class UpscalingTaskDetails(BaseModel):
+#     label: str
+#     service: ServiceDetails
+#     parameters: ServiceParameters
+#     jobs: List[ProcessingJob] = []
+
+
+# class UpscalingTask(UpscalingTaskSummary, UpscalingTaskDetails):
+#     pass
+
+
+class JobsStatusResponse(BaseModel):
+    upscalingTasks: List[UpscalingTaskSummary] = []
+    processingJobs: List[ProcessingJobSummary]= []
+
+
+class BaseJobRequest(BaseModel):
+    title: str
+    label: ProcessType
+    service: ServiceDetails
+    parameters: dict
+
+
+# class UpscaleRequest(BaseJobRequest):
+#     tiles: List[Tile] = []
diff --git a/app/services/processing.py b/app/services/processing.py
@@ -0,0 +1,17 @@
+
+import logging
+from app.platforms.dispatcher import get_processing_platform
+from app.schemas import BaseJobRequest, ProcessingJobSummary 
+
+
+logger = logging.getLogger(__name__)
+
+def create_processing_job(summary: BaseJobRequest) -> ProcessingJobSummary:
+    logger.info(f"Creating processing job with summary: {summary}")
+    
+    platform = get_processing_platform(summary.label)
+    
+    return platform.execute_job(
+        service_id=summary.service.id,
+        parameters=summary.parameters
+    )
diff --git a/pytest b/pytest
@@ -0,0 +1,5 @@
+
+To run the tests, use the following command:
+```bash
+pytest tests/
+```
diff --git a/pytest.ini b/pytest.ini
diff --git a/requirements.txt b/requirements.txt
diff --git a/tests/conftest.py b/tests/conftest.py
diff --git a/tests/test_health.py b/tests/test_health.py
