[feat] add uuid session logger, add typing hints on function arguments

app_gradio_fastapi/helpers/formatters.py

@@ -1,2 +1,30 @@
-def request_formatter(text: str) -> str:
-    return f"transformed {text}."
+import logging
+from fastapi import HTTPException
+
+from app_gradio_fastapi.helpers import session_logger
+
+
+def divide(a: int, b: int) -> float:
+    logging.info(f"a:{a}, b:{b}.")
+    result = a / b
+    logging.info(f"result:{result}.")
+    return result
+
+
+@session_logger.set_uuid_logging
+def request_formatter(text: str) -> dict:
+    logging.info("start request...")
+    try:
+        logging.info(f"input text we need to treat as an integer: {text}.")
+        b = int(text)
+        transformed_text = f"input after integer cast: {b}."
+        try:
+            result_division = divide(100, b)
+            logging.info(f"some_function, result_division:{result_division}.")
+            return {"text": transformed_text, "result": result_division}
+        except ZeroDivisionError as zde:
+            logging.info(f"exception:{zde}.")
+            raise HTTPException(status_code=500, detail="Internal server error")
+    except ValueError as ve:
+        logging.info(f"exception:{ve}.")
+        raise HTTPException(status_code=500, detail="Internal server error")

app_gradio_fastapi/helpers/session_logger.py

@@ -0,0 +1,36 @@
+import contextvars
+import logging
+from functools import wraps
+from typing import Callable
+
+logging_uuid = contextvars.ContextVar("uuid")
+formatter = '%(asctime)s | %(uuid)s [%(pathname)s:%(module)s %(lineno)d] %(levelname)s | %(message)s'
+
+
+loggingType = logging.CRITICAL | logging.ERROR | logging.WARNING | logging.INFO | logging.DEBUG
+
+
+def change_logging(level_log: loggingType = logging.INFO) -> None:
+    old_factory = logging.getLogRecordFactory()
+
+    def record_factory(*args, **kwargs):
+        record = old_factory(*args, **kwargs)
+        record.uuid = logging_uuid.get("uuid")
+        if isinstance(record.msg, str):
+            record.msg = record.msg.replace("\\", "\\\\").replace("\n", "\\n")
+        return record
+
+    logging.setLogRecordFactory(record_factory)
+    logging.basicConfig(level=level_log, format=formatter, force=True)
+
+
+def set_uuid_logging(func: Callable) -> Callable:
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        import uuid
+
+        current_uuid = f"{uuid.uuid4()}"
+        logging_uuid.set(current_uuid)
+        return func(*args, **kwargs)
+
+    return wrapper

app_gradio_fastapi/main.py

@@ -3,8 +3,11 @@ from fastapi import FastAPI
 
 from app_gradio_fastapi import routes
 from app_gradio_fastapi.helpers.formatters import request_formatter
+from app_gradio_fastapi.helpers.session_logger import change_logging
 
 
+change_logging()
+
 CUSTOM_GRADIO_PATH = "/"
 app = FastAPI(title="logging_app", version="1.0")
 app.include_router(routes.router)

app_gradio_fastapi/routes.py

@@ -3,12 +3,15 @@ import logging
 
 from fastapi import APIRouter
 
+from app_gradio_fastapi.helpers import session_logger
+
 
 router = APIRouter()
 
 
 @router.get("/health")
-def health():
+@session_logger.set_uuid_logging
+def health() -> str:
     try:
         logging.info("health check")
         return json.dumps({"msg": "ok"})