Add mechanism to intercept third party std logging

Author: GDYendell

src/fastcs/logging/__init__.py

@@ -1,3 +1,5 @@
+import logging
+
 from loguru import logger as _logger
 
 from ._graylog import GraylogEndpoint as GraylogEndpoint
@@ -84,3 +86,28 @@ def configure_logging(
 
 # Configure logger with defaults - INFO level and disabled
 configure_logging()
+
+
+class _StdLoggingInterceptHandler(logging.Handler):
+    """A std logging handler to forward messages to loguru with nice formatting."""
+
+    def emit(self, record: logging.LogRecord):
+        logger.bind(logger_name=self.name).log(record.levelname, record.getMessage())
+
+
+def intercept_std_logger(logger_name: str):
+    """Intercept std logging messages from the given logger
+
+    To find the correct ``logger_name`` for a message. Add a breakpoint in
+    ``logging.Logger.callHandlers``, debug and run until the log message appears as
+    ``record.msg``. The logger name will be in ``self.name``.
+
+    :param logger_name: Name of the logger to intercept
+
+    """
+    handler = _StdLoggingInterceptHandler()
+    handler.set_name(logger_name)
+
+    std_logger = logging.getLogger(logger_name)
+    std_logger.handlers = [handler]
+    std_logger.propagate = False

src/fastcs/transport/graphql/graphql.py

@@ -10,6 +10,7 @@
 from fastcs.attributes import AttrR, AttrRW, AttrW, T
 from fastcs.controller_api import ControllerAPI
 from fastcs.exceptions import FastCSError
+from fastcs.logging import intercept_std_logger
 
 from .options import GraphQLServerOptions
 
@@ -41,6 +42,10 @@ async def serve(self, options: GraphQLServerOptions | None = None) -> None:
                 log_level=options.log_level,
             )
         )
+        intercept_std_logger("uvicorn.error")
+        intercept_std_logger("uvicorn.access")
+        intercept_std_logger("uvicorn.asgi")
+
         await self._server.serve()
 
 

src/fastcs/transport/rest/rest.py

@@ -8,6 +8,7 @@
 from fastcs.attributes import AttrR, AttrRW, AttrW, T
 from fastcs.controller_api import ControllerAPI
 from fastcs.cs_methods import CommandCallback
+from fastcs.logging import intercept_std_logger
 
 from .options import RestServerOptions
 from .util import (
@@ -44,6 +45,10 @@ async def serve(self, options: RestServerOptions | None):
                 log_level=options.log_level,
             )
         )
+        intercept_std_logger("uvicorn.error")
+        intercept_std_logger("uvicorn.access")
+        intercept_std_logger("uvicorn.asgi")
+
         await self._server.serve()