Improved logging (#8)

* feat: updated logging to use loguru

* fix: duplicate fetching of processing jobs

* feat: improved logging with correlation ID

* fix: linting issues

Author: JanssenBrm

app/config/logger.py

@@ -1,36 +1,86 @@
 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
-        "app.routers": {"level": "DEBUG"},  # all your routers
-        "app.services": {"level": "DEBUG"},  # all your services
-        "app.platforms": {"level": "DEBUG"},  # all platform implementations
-    },
-}
+import os
+import sys
+
+
+from app.middleware.correlation_id import correlation_id_ctx
+from loguru import logger
+
+
+class InterceptHandler(logging.Handler):
+    """
+    Redirect standard logging (incl. uvicorn) to Loguru.
+    """
+
+    def emit(self, record):
+        try:
+            level = logger.level(record.levelname).name
+            corr_id = correlation_id_ctx.get()
+        except ValueError:
+            level = record.levelno
+        except LookupError:
+            corr_id = None
+
+        frame, depth = logging.currentframe(), 2
+        while frame and frame.f_code.co_filename == logging.__file__:
+            frame = frame.f_back
+            depth += 1
+        logger.opt(depth=depth, exception=record.exc_info).log(
+            level, record.getMessage()
+        )
+        if corr_id:
+            logger.bind(correlation_id=corr_id)
+
+
+def correlation_id_filter(record):
+    # Always inject the current correlation ID into the log record
+    record["extra"]["correlation_id"] = correlation_id_ctx.get()
+    return True
 
 
 def setup_logging():
-    logging.config.dictConfig(LOGGING_CONFIG)
+    logger.remove()  # remove default handler
+    env = os.getenv("APP_ENV", "development")
+
+    logger.configure(extra={"correlation_id": None})
+    if env == "production":
+        # JSON logs for ELK
+        logger.add(
+            sys.stdout,
+            serialize=True,
+            backtrace=False,
+            diagnose=False,
+            level="INFO",
+            filter=correlation_id_filter,
+        )
+    else:
+        # Pretty logs for dev
+        logger.add(
+            sys.stdout,
+            colorize=True,
+            format=(
+                "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
+                "<i>{extra[correlation_id]}</i> | "
+                "<level>{level: <8}</level> | "
+                "<cyan>{name}</cyan>:<cyan>{function}</cyan> :<cyan>{line}</cyan> - "
+                "{message}"
+            ),
+            backtrace=True,
+            diagnose=True,
+            level="DEBUG",
+            filter=correlation_id_filter,
+        )
+
+    for name in (
+        "uvicorn",
+        "uvicorn.error",
+        "uvicorn.access",
+        "fastapi",
+        "app.routers",
+        "app.services",
+        "app.platforms",
+    ):
+        logging.getLogger(name).handlers = [InterceptHandler()]
+        logging.getLogger(name).propagate = False
+
+    return logger

app/database/db.py

@@ -1,13 +1,12 @@
-import logging
 import os
 from typing import Optional
 
 from dotenv import load_dotenv
+from loguru import logger
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, declarative_base
 
 load_dotenv()
-logger = logging.getLogger(__name__)
 
 DATABASE_URL: Optional[str] = os.getenv("DATABASE_URL")
 
@@ -36,6 +35,7 @@ def get_db():
         yield db
         db.commit()
     except Exception:
+        logger.exception("An error occurred during database retrieval")
         db.rollback()
         raise
     finally:

app/database/models/processing_job.py

@@ -1,16 +1,13 @@
 import datetime
-import logging
 from typing import List, Optional
+from loguru import logger
 from sqlalchemy import DateTime, Enum, Integer, String
 from app.database.db import Base
 from sqlalchemy.orm import Session, Mapped, mapped_column
 
 from app.schemas.unit_job import ProcessTypeEnum, ProcessingStatusEnum
 
 
-logger = logging.getLogger(__name__)
-
-
 class ProcessingJobRecord(Base):
     __tablename__ = "processing_jobs"
 

app/main.py

@@ -1,5 +1,6 @@
 from fastapi import FastAPI
 
+from app.middleware.correlation_id import add_correlation_id
 from app.platforms.dispatcher import load_processing_platforms
 from .config.logger import setup_logging
 from .config.settings import settings
@@ -15,6 +16,8 @@
     version="1.0.0",
 )
 
+app.middleware("http")(add_correlation_id)
+
 # Register Keycloak - must be done after FastAPI app creation
 # keycloak.register(app, prefix="/auth")  # mounts OIDC endpoints for login if needed
 

app/middleware/correlation_id.py

@@ -0,0 +1,33 @@
+import uuid
+import contextvars
+from fastapi import Request
+from loguru import logger
+
+# Context var to store correlation ID per request
+correlation_id_ctx = contextvars.ContextVar("correlation_id", default="-")
+
+
+async def add_correlation_id(request: Request, call_next):
+    # Try to reuse correlation ID from client headers, else generate one
+    corr_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
+    token = correlation_id_ctx.set(corr_id)
+
+    # Log at start
+    logger.bind(correlation_id=corr_id).info(
+        f"Incoming request: {request.method} - {request.url.path}",
+        method=request.method,
+        path=request.url.path,
+    )
+
+    response = await call_next(request)
+
+    # Add correlation ID to response headers
+    response.headers["X-Correlation-ID"] = corr_id
+
+    # Log at end
+    logger.bind(correlation_id=corr_id).info(
+        f"Request completed with status code {response.status_code}",
+        status_code=response.status_code,
+    )
+    correlation_id_ctx.reset(token)  # Reset after request
+    return response

app/platforms/dispatcher.py

@@ -1,5 +1,6 @@
 import importlib
-import logging
+
+from loguru import logger
 import app.platforms.implementations
 import pkgutil
 from typing import Dict, Type
@@ -8,8 +9,6 @@
 
 PROCESSING_PLATFORMS: Dict[ProcessTypeEnum, Type[BaseProcessingPlatform]] = {}
 
-logger = logging.getLogger(__name__)
-
 
 def register_platform(service_type: ProcessTypeEnum):
     def decorator(cls: Type[BaseProcessingPlatform]):
@@ -40,4 +39,5 @@ def get_processing_platform(service_type: ProcessTypeEnum) -> BaseProcessingPlat
     try:
         return PROCESSING_PLATFORMS[service_type]()
     except KeyError:
+        logger.error(f"Processing platform for service type {service_type} not found.")
         raise ValueError(f"Unsupported service type: {service_type}")

app/platforms/implementations/ogc_api_process.py

@@ -1,12 +1,8 @@
-import logging
-
 from app.platforms.base import BaseProcessingPlatform
 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):

app/platforms/implementations/openeo.py

@@ -1,10 +1,10 @@
 import datetime
-import logging
 import os
 import re
 import urllib
 import jwt
 
+from loguru import logger
 import openeo
 import requests
 from dotenv import load_dotenv
@@ -15,7 +15,6 @@
 from app.schemas.unit_job import ServiceDetails
 
 load_dotenv()
-logger = logging.getLogger(__name__)
 
 # Constants
 BACKEND_AUTH_ENV_MAP = {
@@ -56,7 +55,7 @@ def _connection_expired(self, connection: openeo.Connection) -> bool:
                     logger.debug("JWT bearer token is valid.")
                     return False  # Token is valid
             except Exception as e:
-                logger.warning(f"JWT token validation failed: {e}")
+                logger.error(f"JWT token validation failed: {e}")
                 return True  # Token is expired or invalid
         else:
             logger.warning("No JWT bearer token found in connection.")

app/routers/jobs_status.py

@@ -1,15 +1,14 @@
 import asyncio
-import logging
 
 from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
 from sqlalchemy.orm import Session
+from loguru import logger
 
 from app.database.db import get_db
 from app.schemas.jobs_status import JobsStatusResponse
 from app.services.processing import get_processing_jobs_by_user_id
 
 router = APIRouter()
-logger = logging.getLogger(__name__)
 
 
 @router.get(
@@ -25,8 +24,6 @@ async def get_jobs_status(
     Return combined list of upscaling tasks and processing jobs for the authenticated user.
     """
     logger.debug(f"Fetching jobs list for user {user}")
-    processing_jobs = get_processing_jobs_by_user_id(db, user)
-    print(processing_jobs)
     return JobsStatusResponse(
         upscaling_tasks=[], processing_jobs=get_processing_jobs_by_user_id(db, user)
     )

app/routers/tiles.py

@@ -1,13 +1,12 @@
-import logging
 from fastapi import APIRouter, HTTPException, status
 from geojson_pydantic import GeometryCollection
+from loguru import logger
 
 from app.schemas.tiles import TileRequest
 from app.services.tiles.base import split_polygon_by_grid
 
 
 router = APIRouter()
-logger = logging.getLogger(__name__)
 
 
 @router.post(