feat: change logger library to structlog

.gitignore

@@ -4,6 +4,7 @@ __pycache__
 *.pyc
 flagged/*
 .DS_Store
+.env*
 
 # Created by https://www.toptal.com/developers/gitignore/api/pycharm+all
 # Edit at https://www.toptal.com/developers/gitignore?templates=pycharm+all

app.py

@@ -1,34 +1,96 @@
 import logging
+import os
+import time
 
 import gradio as gr
+import structlog
 import uvicorn
-from fastapi import FastAPI
+from asgi_correlation_id import CorrelationIdMiddleware
+from asgi_correlation_id.context import correlation_id
+from fastapi import FastAPI, Request, Response, APIRouter
+from uvicorn.protocols.utils import get_path_with_query_string
 
-import routes
-from helpers import formatters, session_logger
+from helpers.structlog_setup import setup_logging
+from helpers import formatters
+from dotenv import load_dotenv
 
+from routes import router
 
-session_logger.change_logging()
-app = FastAPI(title="fastapi_with_gradio...", version="1.0")
-logging.info("FastAPI app created, including routes...")
-app.include_router(routes.router)
-logging.info("routes included, creating gradio app")
+load_dotenv()
+
+# LOG_JSON_FORMAT = parse_obj_as(bool, os.getenv("LOG_JSON_FORMAT", False))
+LOG_JSON_FORMAT = bool(os.getenv("LOG_JSON_FORMAT", False))
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
+setup_logging(json_logs=LOG_JSON_FORMAT, log_level=LOG_LEVEL)
+
+access_logger = structlog.stdlib.get_logger(__name__)
+
+app = FastAPI(title="Example API", version="1.0.0")
+
+
+@app.middleware("http")
+async def logging_middleware(request: Request, call_next) -> Response:
+    structlog.contextvars.clear_contextvars()
+    # These context vars will be added to all log entries emitted during the request
+    request_id = correlation_id.get()
+    # print(f"request_id:{request_id}.")
+    structlog.contextvars.bind_contextvars(request_id=request_id)
+
+    start_time = time.perf_counter_ns()
+    # If the call_next raises an error, we still want to return our own 500 response,
+    # so we can add headers to it (process time, request ID...)
+    response = Response(status_code=500)
+    try:
+        response = await call_next(request)
+    except Exception:
+        # TODO: Validate that we don't swallow exceptions (unit test?)
+        structlog.stdlib.get_logger("api.error").exception("Uncaught exception")
+        raise
+    finally:
+        process_time = time.perf_counter_ns() - start_time
+        status_code = response.status_code
+        url = get_path_with_query_string(request.scope)
+        client_host = request.client.host
+        client_port = request.client.port
+        http_method = request.method
+        http_version = request.scope["http_version"]
+        # Recreate the Uvicorn access log format, but add all parameters as structured information
+        access_logger.info(
+            f"""{client_host}:{client_port} - "{http_method} {url} HTTP/{http_version}" {status_code}""",
+            http={
+                "url": str(request.url),
+                "status_code": status_code,
+                "method": http_method,
+                "request_id": request_id,
+                "version": http_version,
+            },
+            network={"client": {"ip": client_host, "port": client_port}},
+            duration=process_time,
+        )
+        response.headers["X-Process-Time"] = str(process_time / 10 ** 9)
+        return response
 
 
+app.include_router(router)
+logging.info("routes included, creating gradio app")
 CUSTOM_GRADIO_PATH = "/"
+
+
 def get_gradio_app():
     with gr.Blocks() as gradio_app:
+        access_logger.info("start gradio app building...")
         gr.Markdown(
             """
             # Hello World!
-            
+
             Start typing below to _see_ the *output*.
 
             Here a [link](https://huggingface.co/spaces/aletrn/gradio_with_fastapi).
             """
         )
         btn = gr.Button(value="Divide et Impera")
-        text_input = gr.Textbox(lines=1, placeholder="10", label="write an integer to divide 100; 0 will raise ZeroDivisionError")
+        text_input = gr.Textbox(lines=1, placeholder="10",
+                                label="write an integer to divide 100; 0 will raise ZeroDivisionError")
         text_output = gr.Textbox(lines=1, placeholder=None, label="Text Output")
         gr.Examples(
             examples=[19, 1, -7, 0],
@@ -36,7 +98,7 @@ def get_gradio_app():
         )
         btn.click(
             formatters.request_formatter,
-            inputs=[text_input],
+            inputs=[text_input, ],
             outputs=[text_output]
         )
     return gradio_app
@@ -44,9 +106,13 @@ def get_gradio_app():
 
 logging.info("mounting gradio app within FastAPI...")
 gradio_app_md = get_gradio_app()
+app.add_middleware(CorrelationIdMiddleware)
 app = gr.mount_gradio_app(app, gradio_app_md, path=CUSTOM_GRADIO_PATH)
 logging.info("gradio app mounted")
 
-
-if __name__ == '__main__':
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+if __name__ == "__main__":
+    try:
+        uvicorn.run("app:app", host="127.0.0.1", port=7860, log_config=None, reload=True)
+    except Exception as ex:
+        logging.error(f"ex:{ex}.")
+        raise ex

helpers/formatters.py

@@ -1,30 +1,30 @@
-import logging
+import structlog
 from fastapi import HTTPException
 
-from helpers import session_logger
+app_logger = structlog.stdlib.get_logger(__name__)
 
 
 def divide(a: int, b: int) -> float:
-    logging.info(f"a:{a}, b:{b}.")
+    app_logger.info(f"a:{a}, b:{b}.")
     result = a / b
-    logging.info(f"result:{result}.")
+    app_logger.info(f"result:{result}.")
     return result
 
 
-@session_logger.set_uuid_logging
+# @session_logger.set_uuid_logging
 def request_formatter(text: str) -> dict:
-    logging.info("start request...")
+    app_logger.info("start request...")
     try:
-        logging.info(f"input text we need to treat as an integer: {text}.")
+        app_logger.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}.")
+            app_logger.info(f"some_function, result_division:{result_division}.")
             return {"text": transformed_text, "result": result_division}
         except ZeroDivisionError as zde:
-            logging.info(f"exception:{zde}.")
+            app_logger.info(f"exception:{zde}.")
             raise HTTPException(status_code=500, detail="Internal server error")
     except ValueError as ve:
-        logging.info(f"exception:{ve}.")
+        app_logger.info(f"exception:{ve}.")
         raise HTTPException(status_code=500, detail="Internal server error")

helpers/structlog_setup.py

@@ -0,0 +1,129 @@
+import logging
+import sys
+
+import structlog
+from structlog.types import EventDict, Processor
+
+
+# https://github.com/hynek/structlog/issues/35#issuecomment-591321744
+def rename_event_key(_, __, event_dict: EventDict) -> EventDict:
+    """
+    Log entries keep the text message in the `event` field, but Datadog
+    uses the `message` field. This processor moves the value from one field to
+    the other.
+    See https://github.com/hynek/structlog/issues/35#issuecomment-591321744
+    """
+    event_dict["message"] = event_dict.pop("event")
+    return event_dict
+
+
+def drop_color_message_key(_, __, event_dict: EventDict) -> EventDict:
+    """
+    Uvicorn logs the message a second time in the extra `color_message`, but we don't
+    need it. This processor drops the key from the event dict if it exists.
+    """
+    event_dict.pop("color_message", None)
+    return event_dict
+
+
+def setup_logging(json_logs: bool = False, log_level: str = "INFO"):
+    timestamper = structlog.processors.TimeStamper(fmt="iso")
+
+    shared_processors: list[Processor] = [
+        structlog.contextvars.merge_contextvars,
+        structlog.stdlib.add_logger_name,
+        structlog.stdlib.add_log_level,
+        structlog.stdlib.PositionalArgumentsFormatter(),
+        structlog.stdlib.ExtraAdder(),
+        drop_color_message_key,
+        timestamper,
+        structlog.processors.StackInfoRenderer(),
+        # adapted from https://www.structlog.org/en/stable/standard-library.html
+        # If the "exc_info" key in the event dict is either true or a
+        # sys.exc_info() tuple, remove "exc_info" and render the exception
+        # with traceback into the "exception" key.
+        structlog.processors.format_exc_info,
+        # If some value is in bytes, decode it to a Unicode str.
+        structlog.processors.UnicodeDecoder(),
+        # Add callsite parameters.
+        structlog.processors.CallsiteParameterAdder(
+            {
+                structlog.processors.CallsiteParameter.FUNC_NAME,
+                structlog.processors.CallsiteParameter.LINENO,
+            }
+        ),
+        # Render the final event dict as JSON.
+    ]
+
+    if json_logs:
+        # We rename the `event` key to `message` only in JSON logs, as Datadog looks for the
+        # `message` key but the pretty ConsoleRenderer looks for `event`
+        shared_processors.append(rename_event_key)
+        # Format the exception only for JSON logs, as we want to pretty-print them when
+        # using the ConsoleRenderer
+        shared_processors.append(structlog.processors.format_exc_info)
+
+    structlog.configure(
+        processors=shared_processors
+        + [
+            # Prepare event dict for `ProcessorFormatter`.
+            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+        ],
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        cache_logger_on_first_use=True,
+    )
+
+    log_renderer: structlog.types.Processor
+    if json_logs:
+        log_renderer = structlog.processors.JSONRenderer()
+    else:
+        log_renderer = structlog.dev.ConsoleRenderer()
+
+    formatter = structlog.stdlib.ProcessorFormatter(
+        # These run ONLY on `logging` entries that do NOT originate within
+        # structlog.
+        foreign_pre_chain=shared_processors,
+        # These run on ALL entries after the pre_chain is done.
+        processors=[
+            # Remove _record & _from_structlog.
+            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+            log_renderer,
+        ],
+    )
+
+    handler = logging.StreamHandler()
+    # Use OUR `ProcessorFormatter` to format all `logging` entries.
+    handler.setFormatter(formatter)
+    root_logger = logging.getLogger()
+    root_logger.addHandler(handler)
+    root_logger.setLevel(log_level.upper())
+
+    for _log in ["uvicorn", "uvicorn.error"]:
+        # Clear the log handlers for uvicorn loggers, and enable propagation
+        # so the messages are caught by our root logger and formatted correctly
+        # by structlog
+        logging.getLogger(_log).handlers.clear()
+        logging.getLogger(_log).propagate = True
+
+    # Since we re-create the access logs ourselves, to add all information
+    # in the structured log (see the `logging_middleware` in main.py), we clear
+    # the handlers and prevent the logs to propagate to a logger higher up in the
+    # hierarchy (effectively rendering them silent).
+    logging.getLogger("uvicorn.access").handlers.clear()
+    logging.getLogger("uvicorn.access").propagate = False
+
+    def handle_exception(exc_type, exc_value, exc_traceback):
+        """
+        Log any uncaught exception instead of letting it be printed by Python
+        (but leave KeyboardInterrupt untouched to allow users to Ctrl+C to stop)
+        See https://stackoverflow.com/a/16993115/3641865
+        """
+        if issubclass(exc_type, KeyboardInterrupt):
+            sys.__excepthook__(exc_type, exc_value, exc_traceback)
+            return
+
+        root_logger.error(
+            "Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)
+        )
+
+    sys.excepthook = handle_exception

helpers/uvicorn_disable_logging.json

@@ -0,0 +1,40 @@
+{
+    "version": 1,
+    "disable_existing_loggers": false,
+    "formatters": {
+        "default": {
+            "()": "uvicorn.logging.DefaultFormatter",
+            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+        },
+        "access": {
+            "()": "uvicorn.logging.AccessFormatter",
+            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+        }
+    },
+    "handlers": {
+        "default": {
+            "formatter": "default",
+            "class": "logging.NullHandler"
+        },
+        "access": {
+            "formatter": "access",
+            "class": "logging.NullHandler"
+        }
+    },
+    "loggers": {
+        "uvicorn.error": {
+            "level": "INFO",
+            "handlers": [
+                "default"
+            ],
+            "propagate": false
+        },
+        "uvicorn.access": {
+            "level": "INFO",
+            "handlers": [
+                "access"
+            ],
+            "propagate": false
+        }
+    }
+}

requirements.txt

@@ -1,2 +1,5 @@
 fastapi==0.110.2
 gradio==4.32.2
+python-dotenv==1.0.1
+structlog==24.4.0
+uvicorn==0.30.1

routes.py

@@ -1,20 +1,17 @@
-import json
-import logging
+import random
 
+import structlog
 from fastapi import APIRouter
 
-from helpers import session_logger
+from helpers import formatters
 
 
+custom_structlog_logger = structlog.stdlib.get_logger(__name__)
 router = APIRouter()
 
 
 @router.get("/health")
-@session_logger.set_uuid_logging
-def health() -> str:
-    try:
-        logging.info("health check")
-        return json.dumps({"msg": "still alive..."})
-    except Exception as e:
-        logging.error(f"exception:{e}.")
-        return json.dumps({"msg": "request failed"})
+def health():
+    n = random.randint(-1, 1)
+    custom_structlog_logger.info(f"random int:{n}")
+    return formatters.request_formatter(f"{n}")