multi job with better matching

Author: suvanbanerjee

.env.example

@@ -0,0 +1,2 @@
+- make the layers rout without the pred name
+- auto0 ofc 

TODO

@@ -1,5 +1,13 @@
 """Services for cloudcasting_backend."""
 
-from cloudcasting_backend.services.s3_downloader import start_download_thread
+from cloudcasting_backend.services.s3_downloader import (
+    run_update_job,
+    get_download_status,
+    get_current_forecast_folder,
+)
 
-__all__ = ["start_download_thread"]
+__all__ = [
+    "run_update_job",
+    "get_download_status",
+    "get_current_forecast_folder",
+]

cloudcasting_backend/services/__init__.py

@@ -64,6 +64,7 @@ class Settings(BaseSettings):
         env_file=".env",
         env_prefix="CLOUDCASTING_BACKEND_",
         env_file_encoding="utf-8",
+        extra="ignore",  # Ignore extra fields from .env file
     )
 
 

cloudcasting_backend/services/s3_downloader.py

@@ -2,16 +2,16 @@
 
 import os
 from pathlib import Path
-from typing import List
+from typing import List, Optional
 
 from fastapi import APIRouter, HTTPException, status
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
 
 from cloudcasting_backend.services.s3_downloader import (
-    download_s3_folder,
+    run_update_job,
+    get_download_status,
     get_current_forecast_folder,
-    convert_zarr_to_geotiffs,
     GEOTIFF_STORAGE_PATH,
 )
 from cloudcasting_backend.settings import settings
@@ -33,6 +33,22 @@ class AvailableLayersResponse(BaseModel):
     steps: List[int]
 
 
+class BackgroundTaskResponse(BaseModel):
+    """Response model for background tasks."""
+    
+    task_id: str
+    message: str
+
+
+class DownloadStatusResponse(BaseModel):
+    """Response model for download status."""
+    
+    is_running: bool
+    current_task: Optional[str] = None
+    last_completed: Optional[str] = None
+    error: Optional[str] = None
+
+
 @router.get("/status", response_model=CloudcastingResponse)
 async def get_cloudcasting_status() -> CloudcastingResponse:
     """
@@ -61,48 +77,46 @@ async def get_cloudcasting_status() -> CloudcastingResponse:
     )
 
 
-@router.post("/trigger-download", response_model=CloudcastingResponse)
-async def trigger_download() -> CloudcastingResponse:
+@router.post("/trigger-download", response_model=BackgroundTaskResponse)
+async def trigger_download() -> BackgroundTaskResponse:
     """
-    Manually trigger a download of the latest cloudcasting data.
-    Downloads the data and converts it to GeoTIFF format, then completes.
+    Trigger a background download of the latest cloudcasting data.
+    Downloads the data and converts it to GeoTIFF format in the background.
+    Returns immediately with a task ID that can be used to check status.
     """
     try:
-        bucket_name = settings.s3_bucket_name
-        s3_folder = get_current_forecast_folder()
-        local_dir = settings.zarr_storage_path
-
-        # Ensure directory exists
-        Path(local_dir).mkdir(parents=True, exist_ok=True)
-
-        # Download the data
-        downloaded_zarr_path = download_s3_folder(bucket_name, s3_folder, local_dir)
-
-        # Convert to GeoTIFF if download was successful
-        if downloaded_zarr_path:
-            convert_zarr_to_geotiffs(downloaded_zarr_path, GEOTIFF_STORAGE_PATH)
-
-        # Check if data is available and return path
-        latest_path = Path(local_dir) / "cloudcasting_forecast" / "latest.zarr"
-        if latest_path.exists():
-            relative_path = latest_path.relative_to(
-                Path(settings.zarr_storage_path).parent,
-            )
-            data_path = f"/static/{relative_path}"
-        else:
-            data_path = ""
-
-        return CloudcastingResponse(
-            message="Download and conversion completed successfully",
-            data_path=data_path,
+        # Directly call run_update_job to start the background process
+        run_update_job()
+        
+        # Get the current status to return task info
+        status_info = get_download_status()
+        task_id = status_info.get("current_task", "unknown")
+        
+        return BackgroundTaskResponse(
+            task_id=task_id,
+            message=f"Background download started: {task_id}"
+        )
+    except RuntimeError as e:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(e),
         )
     except Exception as e:
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to download data: {e!s}",
+            detail=f"Failed to start download: {e!s}",
         )
 
 
+@router.get("/download-status", response_model=DownloadStatusResponse)
+async def get_download_status_endpoint() -> DownloadStatusResponse:
+    """
+    Get the current status of background download processes.
+    """
+    status_info = get_download_status()
+    return DownloadStatusResponse(**status_info)
+
+
 @router.get("/layers", response_model=AvailableLayersResponse)
 async def get_available_layers() -> AvailableLayersResponse:
     """

cloudcasting_backend/settings.py

@@ -1,7 +1,9 @@
+import os
 from contextlib import asynccontextmanager
 from typing import AsyncGenerator
 
 from fastapi import FastAPI
+from loguru import logger as log
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.sdk.resources import (
@@ -14,7 +16,6 @@
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from opentelemetry.trace import set_tracer_provider
 
-from cloudcasting_backend.services import start_download_thread
 from cloudcasting_backend.settings import settings
 
 
@@ -93,10 +94,6 @@ async def lifespan_setup(
     setup_opentelemetry(app)
     app.middleware_stack = app.build_middleware_stack()
 
-    # Start the scheduled S3 download thread
-    # Disabled: Auto-running thread removed, use /trigger-download endpoint instead
-    # if settings.environment != "pytest":  # Don't start in test environment
-    #     start_download_thread()
 
     yield
     stop_opentelemetry(app)