[feat] reformat code, log context request id

dockerfiles/dockerfile-base-webserver

@@ -4,7 +4,7 @@ WORKDIR /code
 
 RUN which python
 RUN python --version
-RUN python -m pip install --no-cache-dir fastapi uvicorn
+RUN python -m pip install --no-cache-dir fastapi uvicorn loguru
 
 RUN useradd -m -u 1000 user
 

requirements.txt

@@ -1,5 +1,6 @@
 fastapi
 bson
+loguru
 python-dotenv
 segment-geospatial
 uvicorn[standard]

src/__init__.py

@@ -0,0 +1,4 @@
+from src.utilities.utilities import setup_logging
+
+
+app_logger = setup_logging(debug=True)

src/main.py

@@ -1,18 +1,38 @@
 import json
-from typing import List
+import uuid
+
 from fastapi import FastAPI, HTTPException, Request, status
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel
 
+from src import app_logger
 from src.utilities.type_hints import input_floatlist, input_floatlist2
-from src.utilities.utilities import setup_logging
 
 
 app = FastAPI()
 
 
+@app.middleware("http")
+async def request_middleware(request, call_next):
+    request_id = str(uuid.uuid4())
+    with app_logger.contextualize(request_id=request_id):
+        app_logger.info("Request started")
+
+        try:
+            response = await call_next(request)
+
+        except Exception as ex:
+            app_logger.error(f"Request failed: {ex}")
+            response = JSONResponse(content={"success": False}, status_code=500)
+
+        finally:
+            response.headers["X-Request-ID"] = request_id
+            app_logger.info(f"Request ended")
+            return response
+
+
 class Input(BaseModel):
     x1: float
     y1: float
@@ -39,6 +59,8 @@ def get_parsed_bbox_points(request_input: Input) -> BBoxWithPoint:
 
 @app.post("/post_test")
 async def post_test(request_input: Input) -> JSONResponse:
+    request_body = get_parsed_bbox_points(request_input)
+    app_logger.info(f"request_body:{request_body}.")
     return JSONResponse(
         status_code=200,
         content=get_parsed_bbox_points(request_input)
@@ -47,6 +69,7 @@ async def post_test(request_input: Input) -> JSONResponse:
 
 @app.get("/hello")
 async def hello() -> JSONResponse:
+    app_logger.info(f"hello")
     return JSONResponse(status_code=200, content={"msg": "hello"})
 
 
@@ -56,51 +79,41 @@ def samgeo(request_input: Input):
 
     from src.prediction_api.predictor import base_predict
 
-    local_logger = setup_logging()
-    local_logger.info("starting inference request...")
+    app_logger.info("starting inference request...")
 
     try:
         import time
 
         time_start_run = time.time()
-        # debug = True
-        # local_logger = setup_logging(debug)
         request_body = get_parsed_bbox_points(request_input)
-        local_logger.info(f"request_body:{request_body}.")
+        app_logger.info(f"request_body:{request_body}.")
         try:
             output = base_predict(
                 bbox=request_body["bbox"],
                 point_coords=request_body["point"]
             )
-
-            duration_run = time.time() - time_start_run
-            body = {
-                "duration_run": duration_run,
-                # "request_id": request_id
-            }
-            local_logger.info(f"body:{body}.")
+            body = {"duration_run": time.time() - time_start_run}
+            app_logger.info(f"body:{body}.")
             body["output"] = output
-            # local_logger.info(f"End_request::{request_id}...")
             return JSONResponse(status_code=200, content={"body": json.dumps(body)})
         except Exception as inference_exception:
             home_content = subprocess.run("ls -l /home/user", shell=True, universal_newlines=True, stdout=subprocess.PIPE)
-            local_logger.error(f"/home/user ls -l: {home_content.stdout}.")
-            local_logger.error(f"inference error:{inference_exception}.")
+            app_logger.error(f"/home/user ls -l: {home_content.stdout}.")
+            app_logger.error(f"inference error:{inference_exception}.")
             return HTTPException(status_code=500, detail="Internal server error on inference")
     except Exception as generic_exception:
-        local_logger.error(f"generic error:{generic_exception}.")
+        app_logger.error(f"generic error:{generic_exception}.")
         return HTTPException(status_code=500, detail="Generic internal server error")
 
 
 @app.exception_handler(RequestValidationError)
 async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
-    local_logger = setup_logging()
-    local_logger.error(f"exception errors: {exc.errors()}.")
-    local_logger.error(f"exception body: {exc.body}.")
+    app_logger.error(f"exception errors: {exc.errors()}.")
+    app_logger.error(f"exception body: {exc.body}.")
     headers = request.headers.items()
-    local_logger.error(f'request header: {dict(headers)}.' )
+    app_logger.error(f'request header: {dict(headers)}.' )
     params = request.query_params.items()
-    local_logger.error(f'request query params: {dict(params)}.')
+    app_logger.error(f'request query params: {dict(params)}.')
     return JSONResponse(
         status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
         content={"msg": "Error - Unprocessable Entity"}
@@ -109,12 +122,11 @@ async def request_validation_exception_handler(request: Request, exc: RequestVal
 
 @app.exception_handler(HTTPException)
 async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
-    local_logger = setup_logging()
-    local_logger.error(f"exception: {str(exc)}.")
+    app_logger.error(f"exception: {str(exc)}.")
     headers = request.headers.items()
-    local_logger.error(f'request header: {dict(headers)}.' )
+    app_logger.error(f'request header: {dict(headers)}.' )
     params = request.query_params.items()
-    local_logger.error(f'request query params: {dict(params)}.')
+    app_logger.error(f'request query params: {dict(params)}.')
     return JSONResponse(
         status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
         content={"msg": "Error - Internal Server Error"}

src/prediction_api/predictor.py

@@ -1,11 +1,9 @@
 # Press the green button in the gutter to run the script.
 import json
 
+from src import app_logger
 from src.utilities.constants import ROOT
 from src.utilities.type_hints import input_floatlist, input_floatlist2
-from src.utilities.utilities import setup_logging
-
-local_logger = setup_logging()
 
 
 def base_predict(
@@ -14,13 +12,13 @@ def base_predict(
     from samgeo import SamGeo, tms_to_geotiff
 
     image = f"{root_folder}/satellite.tif"
-    local_logger.info(f"start tms_to_geotiff using bbox:{bbox}, type:{type(bbox)}.")
+    app_logger.info(f"start tms_to_geotiff using bbox:{bbox}, type:{type(bbox)}.")
     for coord in bbox:
-        local_logger.info(f"coord:{coord}, type:{type(coord)}.")
+        app_logger.info(f"coord:{coord}, type:{type(coord)}.")
     # bbox: image input coordinate
     tms_to_geotiff(output=image, bbox=bbox, zoom=zoom, source="Satellite", overwrite=True)
 
-    local_logger.info(f"geotiff created, start to initialize samgeo instance (read model {model_name} from {root_folder})...")
+    app_logger.info(f"geotiff created, start to initialize samgeo instance (read model {model_name} from {root_folder})...")
 
     predictor = SamGeo(
         model_type=model_name,
@@ -28,21 +26,22 @@ def base_predict(
         automatic=False,
         sam_kwargs=None,
     )
-    local_logger.info(f"initialized samgeo instance, start to set_image {image}...")
+    app_logger.info(f"initialized samgeo instance, start to set_image {image}...")
     predictor.set_image(image)
     output_name = f"{root_folder}/output.tif"
 
-    local_logger.info(f"done set_image, start prediction...")
+    app_logger.info(f"done set_image, start prediction...")
     predictor.predict(point_coords, point_labels=len(point_coords), point_crs=point_crs, output=output_name)
 
-    local_logger.info(f"done prediction, start tiff to geojson conversion...")
+    app_logger.info(f"done prediction, start tiff to geojson conversion...")
 
     # geotiff to geojson
     vector = f"{root_folder}/feats.geojson"
     predictor.tiff_to_geojson(output_name, vector, bidx=1)
-    local_logger.info(f"start reading geojson...")
+    app_logger.info(f"start reading geojson...")
 
     with open(vector) as out_gdf:
-        out_gdf_str = json.load(out_gdf)
-        local_logger.info(f"number of fields in geojson output:{len(out_gdf_str)}.")
-        return out_gdf_str
+        out_gdf_str = out_gdf.read()
+        out_gdf_json = json.loads(out_gdf_str)
+        app_logger.info(f"geojson string output length:{len(out_gdf_str)}.")
+        return out_gdf_json

src/utilities/measures.py

@@ -1,76 +0,0 @@
-"""helpers for compute measures: hash, time benchmarks"""
-from pathlib import Path
-
-
-def hash_calculate(arr: any, debug: bool = False) -> str or bytes:
-    """
-    Return computed hash from input variable (typically a numpy array).
-
-    Args:
-        arr: input variable
-        debug: logging debug argument
-
-    Returns:
-        str or bytes: computed hash from input variable
-
-    """
-    import hashlib
-    import numpy as np
-    from base64 import b64encode
-
-    from src.utilities.utilities import setup_logging
-    local_logger = setup_logging(debug)
-
-    if isinstance(arr, np.ndarray):
-        hash_fn = hashlib.sha256(arr.data)
-    elif isinstance(arr, dict):
-        import json
-        from src.utilities.serialize import serialize
-
-        serialized = serialize(arr)
-        variable_to_hash = json.dumps(serialized, sort_keys=True).encode('utf-8')
-        hash_fn = hashlib.sha256(variable_to_hash)
-    elif isinstance(arr, str):
-        try:
-            hash_fn = hashlib.sha256(arr)
-        except TypeError:
-            local_logger.warning(f"TypeError, re-try encoding arg:{arr},type:{type(arr)}.")
-            hash_fn = hashlib.sha256(arr.encode('utf-8'))
-    elif isinstance(arr, bytes):
-        hash_fn = hashlib.sha256(arr)
-    else:
-        raise ValueError(f"variable 'arr':{arr} not yet handled.")
-    return b64encode(hash_fn.digest())
-
-
-def sha256sum(filename: Path or str) -> str:
-    """
-    Return computed hash for input file.
-
-    Args:
-        filename: input variable
-
-    Returns:
-        str: computed hash
-
-    """
-    import hashlib
-    import mmap
-
-    h = hashlib.sha256()
-    with open(filename, 'rb') as f:
-        with mmap.mmap(f.fileno(), 0, prot=mmap.PROT_READ) as mm:
-            h.update(mm)
-    return h.hexdigest()
-
-
-def perf_counter() -> float:
-    """
-    Performance counter for benchmarking.
-
-    Returns:
-        float: computed time value at execution time
-
-    """
-    import time
-    return time.perf_counter()

src/utilities/serialize.py

@@ -1,86 +0,0 @@
-"""Serialize objects"""
-from typing import Mapping
-
-from src.utilities.type_hints import ts_dict_str3, ts_dict_str2
-
-
-def serialize(obj:any, include_none:bool=False) -> object:
-    """
-    Return the input object into a serializable one
-
-    Args:
-        obj: Object to serialize
-        include_none: bool to indicate if include also keys with None values during dict serialization
-
-    Returns:
-        object: serialized object
-
-    """
-    return _serialize(obj, include_none)
-
-
-def _serialize(obj:any, include_none:bool) -> any:
-    import numpy as np
-
-    primitive = (int, float, str, bool)
-    # print(type(obj))
-    try:
-        if obj is None:
-            return None
-        elif isinstance(obj, np.integer):
-            return int(obj)
-        elif isinstance(obj, np.floating):
-            return float(obj)
-        elif isinstance(obj, np.ndarray):
-            return obj.tolist()
-        elif isinstance(obj, primitive):
-            return obj
-        elif type(obj) is list:
-            return _serialize_list(obj, include_none)
-        elif type(obj) is tuple:
-            return list(obj)
-        elif type(obj) is bytes:
-            return _serialize_bytes(obj)
-        elif isinstance(obj, Exception):
-            return _serialize_exception(obj)
-        # elif isinstance(obj, object):
-        #     return _serialize_object(obj, include_none)
-        else:
-            return _serialize_object(obj, include_none)
-    except Exception as e_serialize:
-        from src.utilities.utilities import setup_logging
-        serialize_logger = setup_logging()
-        serialize_logger.error(f"e_serialize::{e_serialize}, type_obj:{type(obj)}, obj:{obj}.")
-        return f"object_name:{str(obj)}__object_type_str:{str(type(obj))}."
-
-
-def _serialize_object(obj:Mapping[any, object], include_none:bool) -> dict[any]:
-    from bson import ObjectId
-
-    res = {}
-    if type(obj) is not dict:
-        keys = [i for i in obj.__dict__.keys() if (getattr(obj, i) is not None) or include_none]
-    else:
-        keys = [i for i in obj.keys() if (obj[i] is not None) or include_none]
-    for key in keys:
-        if type(obj) is not dict:
-            res[key] = _serialize(getattr(obj, key), include_none)
-        elif isinstance(obj[key], ObjectId):
-            continue
-        else:
-            res[key] = _serialize(obj[key], include_none)
-    return res
-
-
-def _serialize_list(ls:list, include_none:bool) -> list:
-    return [_serialize(elem, include_none) for elem in ls]
-
-
-def _serialize_bytes(b:bytes) -> ts_dict_str2:
-    import base64
-    encoded = base64.b64encode(b)
-    return {"value": encoded.decode('ascii'), "type": "bytes"}
-
-
-def _serialize_exception(e: Exception) -> ts_dict_str3:
-    return {"msg": str(e), "type": str(type(e)), **e.__dict__}

src/utilities/type_hints.py

@@ -1,13 +1,6 @@
 """custom type hints"""
-from typing import List, Dict, Tuple
+from typing import List
 
-import numpy as np
-
-# ts_ddict1, ts_float64_1, ts_float64_2, ts_dict_str3, ts_dict_str2
 input_floatlist = List[float]
 input_floatlist2 = List[input_floatlist]
-ts_ddict1 = Dict[str, Dict[str, any], Dict, Dict, any]
-ts_dict_str2 = Dict[str, str]
-ts_dict_str3 = Dict[str, str, any]
-ts_float64_1 = Tuple[np.float64, np.float64, np.float64, np.float64, np.float64, np.float64]
-ts_float64_2 = Tuple[np.float64, np.float64, np.float64, np.float64, np.float64, np.float64, np.float64]
+

src/utilities/utilities.py

@@ -1,10 +1,10 @@
 """Various utilities (logger, time benchmark, args dump, numerical and stats info)"""
-import logging
+import loguru
 
 from src.utilities.constants import ROOT
 
 
-def setup_logging(debug: bool = False, formatter: str = '%(asctime)s - %(name)s - %(funcName)s(): line %(lineno)d - %(levelname)s - %(message)s') -> logging.Logger:
+def setup_logging(debug: bool = False, formatter: str = "{time} - {level} - ({extra[request_id]}) {message} ") -> loguru.logger:
     """
     Create a logging instance with log string formatter.
 
@@ -16,21 +16,12 @@ def setup_logging(debug: bool = False, formatter: str = '%(asctime)s - %(name)s
         Logger
 
     """
-    import logging
     import sys
 
-    logger = logging.getLogger()
-    for h in logger.handlers:
-        logger.removeHandler(h)
-
-    h = logging.StreamHandler(sys.stdout)
-
-    h.setFormatter(logging.Formatter(formatter))
-    logger.addHandler(h)
-    logger.setLevel(logging.INFO)
-
-    if debug:
-        logger.setLevel(logging.DEBUG)
+    logger = loguru.logger
+    logger.remove()
+    level_logger = "DEBUG" if debug else "INFO"
+    logger.add(sys.stdout, format=formatter, level=level_logger)
     logger.debug(f"type_logger:{type(logger)}.")
     return logger