feat: updated logging to use loguru

Author: JanssenBrm

app/config/logger.py

@@ -1,36 +1,69 @@
 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 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
+        except ValueError:
+            level = record.levelno
+        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()
+        )
 
 
 def setup_logging():
-    logging.config.dictConfig(LOGGING_CONFIG)
+    logger.remove()  # remove default handler
+    env = os.getenv("APP_ENV", "development")
+
+    if env == "production":
+        # JSON logs for ELK
+        logger.add(
+            sys.stdout,
+            serialize=True,
+            backtrace=False,
+            diagnose=False,
+            level="INFO",
+        )
+    else:
+        # Pretty logs for dev
+        logger.add(
+            sys.stdout,
+            colorize=True,
+            format=(
+                "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
+                "<level>{level: <8}</level> | "
+                "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
+                "{message}"
+            ),
+            backtrace=True,
+            diagnose=True,
+            level="DEBUG",
+        )
+
+    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/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,11 +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__)
+from loguru import logger
 
 
 @register_platform(ProcessTypeEnum.OGC_API_PROCESS)

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(
@@ -26,7 +25,6 @@ async def get_jobs_status(
     """
     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(

app/routers/unit_jobs.py

@@ -1,6 +1,5 @@
-import logging
-
 from fastapi import APIRouter, Depends, HTTPException, status
+from loguru import logger
 from sqlalchemy.orm import Session
 
 from app.database.db import get_db
@@ -10,7 +9,6 @@
 # from app.auth import get_current_user
 
 router = APIRouter()
-logger = logging.getLogger(__name__)
 
 
 @router.post(
@@ -26,7 +24,7 @@ async def create_unit_job(
     try:
         return create_processing_job(db, user, payload)
     except Exception as e:
-        logger.error(f"Error creating unit job for user {user}: {e}")
+        logger.exception(f"Error creating unit job for user {user}: {e}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"An error occurred while creating the processing job: {e}",
@@ -43,6 +41,7 @@ async def get_job(
 ) -> ProcessingJob:
     job = get_processing_job_by_user_id(db, job_id, user)
     if not job:
+        logger.error(f"Processing job {job_id} not found for user {user}")
         raise HTTPException(
             status_code=404,
             detail=f"Processing job {job_id} not found",

app/services/processing.py

@@ -1,6 +1,7 @@
 import json
-import logging
 from typing import List, Optional
+
+from loguru import logger
 from app.database.models.processing_job import (
     ProcessingJobRecord,
     get_job_by_user_id,
@@ -21,9 +22,6 @@
 )
 
 
-logger = logging.getLogger(__name__)
-
-
 def create_processing_job(
     database: Session, user: str, summary: BaseJobRequest
 ) -> ProcessingJobSummary: