actually use fastapi lifetime manager to setup logging

.vscode/launch.json

@@ -4,17 +4,21 @@
     // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
     "version": "0.2.0",
     "configurations": [
-        // backend - python app that launches a uvicorn server
+        // backend - python using fastapi
         {
             "name": "Backend - debug",
             "type": "debugpy",
             "request": "launch",
-            "program": "launcher.py",
             "env": {
                 "DEBUG": "true"
             },
             "jinja": true,
-            "cwd": "${workspaceFolder}/backend"
+            "cwd": "${workspaceFolder}/backend",
+            "module": "fastapi",
+            "args": [
+                "dev",
+                "src/main.py"
+            ]
         },
         {
             "name": "Backend - tester",

backend/Dockerfile

@@ -7,7 +7,6 @@ RUN pip install pipenv
 RUN pipenv install --deploy --system
 
 COPY src src
-COPY launcher.py .
 
 EXPOSE 8000
 
@@ -17,4 +16,4 @@ ENV OSM_CACHE_DIR=/cache
 ENV MEMCACHED_HOST_PATH=none
 ENV LOKI_URL=none
 
-CMD ["python", "launcher.py"]
+CMD ["fastapi", "src/main.py", "--port", "8000", "--workers", "$NUM_WORKERS"]

backend/launcher.py

@@ -1,82 +0,0 @@
-"""Launcher for the FastAPI application. Fundametally this replicates the functionality of the uvicorn and fastapi CLI interfaces, but we need this to setup the logging correctly (and most importantly globally)"""
-
-import os
-import uvicorn
-import logging
-
-logger = logging.getLogger(__name__)
-is_debug = os.getenv('DEBUG', "false") == "true"
-is_kubernetes = os.getenv('KUBERNETES_SERVICE_HOST', None) is not None
-
-def logger_setup():
-    """
-    Setup the global logging configuration
-    """
-    logging_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-
-    # Make uvicorn conform to the global logging configuration
-    uvicorn_logger = logging.getLogger('uvicorn')
-    uvicorn_logger.propagate = True
-    uvicorn_logger.handlers = []  # Remove default handlers to avoid duplicate logs
-
-    if is_kubernetes:
-        # in that case we want to log to stdout and also to loki
-        from loki_logger_handler.loki_logger_handler import LokiLoggerHandler
-        loki_url = os.getenv('LOKI_URL')
-        if loki_url is None:
-            raise ValueError("LOKI_URL environment variable is not set")
-        
-        loki_handler = LokiLoggerHandler(
-            url = loki_url,
-            labels = {'app': 'anyway', 'environment': 'staging' if is_debug else 'production'}
-        )
-
-        logger.info(f"Logging to Loki at {loki_url} with {loki_handler.labels} and {is_debug=}")
-        if is_debug:
-            logging.basicConfig(
-                format = logging_format,
-                level = logging.DEBUG,
-                handlers = [loki_handler, logging.StreamHandler()]
-            )
-            # we need to silence the debug logs made by the loki handler
-            logging.getLogger('urllib3.connectionpool').setLevel(logging.INFO)
-        else:
-            logging.basicConfig(
-                format = logging_format,
-                level = logging.INFO,
-                handlers = [loki_handler, logging.StreamHandler()]
-            )
-    else:
-        # if we are in a debug (local) session, set verbose and rich logging
-        from rich.logging import RichHandler
-        logging.basicConfig(
-            format = logging_format,
-            level = logging.DEBUG,
-            handlers = [RichHandler()]
-        )
-
-
-def uvicorn_run():
-    """
-    Run the FastAPI application using uvicorn
-    """
-    num_workers = int(os.getenv('NUM_WORKERS', 1))
-    logger.info(f"Starting FastAPI+uvicorn with {num_workers=}, {is_debug=}, {is_kubernetes=}")
-    uvicorn.run(
-        # we could in theory directly import the app and pass it as an object
-        # this 'import string' is required for hot reloading and scaling
-        'src.main:app',
-        host = '0.0.0.0',
-        port = 8000,
-        log_config = None,
-        access_log = True,
-        # Disable uvicorn's logging configuration so that it inherits the global one
-        workers = num_workers,
-        # Hot reload breaks logging, so we leave it disabled
-        # reload = is_debug
-    )
-    
-
-if __name__ == '__main__':
-    logger_setup()
-    uvicorn_run()

backend/logging_config.yaml

@@ -1,29 +0,0 @@
-version: 1
-disable_existing_loggers: False
-formatters:
-    standard:
-        format: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-handlers:
-    console:
-        class: logging.StreamHandler
-        formatter: standard
-        level: DEBUG
-    rich:
-        class: rich.logging.RichHandler
-        level: DEBUG
-    loki:
-        class: loki_logger_handler.loki_logger_handler.LokiLoggerHandler
-        level: DEBUG
-        formatter: standard
-        url: ${LOKI_URL}
-        labels:
-            app: anyway
-            environment: ${ENVIRONMENT}
-loggers:
-    uvicorn:
-        handlers: [console, rich, loki]
-        level: DEBUG
-        propagate: False
-root:
-    handlers: [console, rich, loki]
-    level: DEBUG

backend/src/logging_config.py

@@ -0,0 +1,56 @@
+"""Sets up global logging configuration for the application."""
+
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+
+def configure_logging():
+    """
+    Called at startup of a FastAPI application instance to setup logging. Depending on the environment, it will log to stdout or to Loki.
+    """
+
+    is_debug = os.getenv('DEBUG', "false") == "true"
+    is_kubernetes = os.getenv('KUBERNETES_SERVICE_HOST') is not None
+
+
+    if not is_kubernetes:
+        # in that case we want to log to stdout and also to loki
+        from loki_logger_handler.loki_logger_handler import LokiLoggerHandler
+        loki_url = os.getenv('LOKI_URL')
+        loki_url = "http://localhost:3100/loki/api/v1/push"
+        if loki_url is None:
+            raise ValueError("LOKI_URL environment variable is not set")
+        
+        loki_handler = LokiLoggerHandler(
+            url = loki_url,
+            labels = {'app': 'anyway', 'environment': 'staging' if is_debug else 'production'}
+        )
+
+        logger.info(f"Logging to Loki at {loki_url} with {loki_handler.labels} and {is_debug=}")
+        logging_handlers = [loki_handler, logging.StreamHandler()]
+        logging_level = logging.DEBUG if is_debug else logging.INFO
+        # no need for time since it's added by loki or can be shown in kube logs
+        logging_format = '%(name)s - %(levelname)s - %(message)s'
+
+    else:
+        # if we are in a debug (local) session, set verbose and rich logging
+        from rich.logging import RichHandler
+        logging_handlers = [RichHandler()]
+        logging_level = logging.DEBUG if is_debug else logging.INFO
+        logging_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+
+
+
+    logging.basicConfig(
+        level = logging_level,
+        format = logging_format,
+        handlers = logging_handlers
+    )
+
+    # also overwrite the uvicorn loggers
+    logging.getLogger('uvicorn').handlers = logging_handlers
+    logging.getLogger('uvicorn.access').handlers = logging_handlers
+    logging.getLogger('uvicorn.error').handlers = logging_handlers
+

backend/src/main.py

@@ -2,7 +2,9 @@
 
 import logging
 from fastapi import FastAPI, HTTPException, Query
+from contextlib import asynccontextmanager
 
+from .logging_config import configure_logging
 from .structs.landmark import Landmark, Toilets
 from .structs.preferences import Preferences
 from .structs.linked_landmarks import LinkedLandmarks
@@ -13,15 +15,26 @@ from .utils.optimizer import Optimizer
 from .utils.refiner import Refiner
 from .cache import client as cache_client
 
-
 logger = logging.getLogger(__name__)
 
-app = FastAPI()
 manager = LandmarkManager()
 optimizer = Optimizer()
 refiner = Refiner(optimizer=optimizer)
 
 
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Function to run at the start of the app"""
+    logger.info("Setting up logging")
+    configure_logging()
+    yield
+    logger.info("Shutting down logging")
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+
 @app.post("/trip/new")
 def new_trip(preferences: Preferences,
              start: tuple[float, float],

backend/src/structs/landmark.py

@@ -1,7 +1,7 @@
 """Definition of the Landmark class to handle visitable objects across the world."""
 
 from typing import Optional, Literal
-from uuid import uuid4
+from uuid import uuid4, UUID
 from pydantic import BaseModel, Field
 
 
@@ -29,12 +29,12 @@ class Landmark(BaseModel) :
         description (Optional[str]): A text description of the landmark.
         duration (Optional[int]): The estimated time to visit the landmark (in minutes).
         name_en (Optional[str]): The English name of the landmark.
-        uuid (str): A unique identifier for the landmark, generated by default using uuid4.
+        uuid (UUID): A unique identifier for the landmark, generated by default using uuid4.
         must_do (Optional[bool]): Whether the landmark is a "must-do" attraction.
         must_avoid (Optional[bool]): Whether the landmark should be avoided.
         is_secondary (Optional[bool]): Whether the landmark is secondary or less important.
         time_to_reach_next (Optional[int]): Estimated time (in minutes) to reach the next landmark.
-        next_uuid (Optional[str]): UUID of the next landmark in sequence (if applicable).
+        next_uuid (Optional[UUID]): UUID of the next landmark in sequence (if applicable).
     """
 
     # Properties of the landmark
@@ -52,7 +52,7 @@ class Landmark(BaseModel) :
     name_en : Optional[str] = None
 
     # Unique ID of a given landmark
-    uuid: str = Field(default_factory=uuid4)
+    uuid: UUID = Field(default_factory=uuid4)
 
     # Additional properties depending on specific tour
     must_do : Optional[bool] = False
@@ -60,7 +60,7 @@ class Landmark(BaseModel) :
     is_secondary : Optional[bool] = False
 
     time_to_reach_next : Optional[int] = 0
-    next_uuid : Optional[str] = None
+    next_uuid : Optional[UUID] = None
 
     def __str__(self) -> str:
         """

backend/src/structs/trip.py

@@ -1,6 +1,6 @@
 """Definition of the Trip class."""
 
-import uuid
+from uuid import uuid4, UUID
 from pydantic import BaseModel, Field
 from pymemcache.client.base import Client
 
@@ -19,9 +19,9 @@ class Trip(BaseModel):
     Methods:
         from_linked_landmarks: create a Trip from LinkedLandmarks object.
     """
-    uuid: str = Field(default_factory=uuid.uuid4)
+    uuid: UUID = Field(default_factory=uuid4)
     total_time: int
-    first_landmark_uuid: str
+    first_landmark_uuid: UUID
 
 
     @classmethod
@@ -31,7 +31,7 @@ class Trip(BaseModel):
         """
         trip = Trip(
             total_time = landmarks.total_time,
-            first_landmark_uuid = str(landmarks[0].uuid)
+            first_landmark_uuid = landmarks[0].uuid
         )
 
         # Store the trip in the cache