452 make uvicorns logs follow the configured format

docs/reference/logger.md

@@ -15,6 +15,7 @@ from fastapi_toolsets.logger import (
     JsonFormatter,
     LoggingConfig,
     LogLevel,
+    LogStream,
     LogStyle,
     RequestLoggingMiddleware,
     SharedRotatingFileHandler,

src/fastapi_toolsets/logger/__init__.py

@@ -13,6 +13,7 @@
     DEFAULT_QUIET_LOGGERS,
     LoggingConfig,
     LogLevel,
+    LogStream,
     LogStyle,
     configure_logging,
 )
@@ -30,6 +31,7 @@
     "ContextFilter",
     "JsonFormatter",
     "LogLevel",
+    "LogStream",
     "LogStyle",
     "LoggingConfig",
     "RequestLoggingMiddleware",

src/fastapi_toolsets/logger/config.py

@@ -6,7 +6,7 @@
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field, replace
 from pathlib import Path
-from typing import Any, Literal
+from typing import Any, Literal, get_args
 
 from .context import ContextFilter, skip_logged_exceptions
 from .files import SharedRotatingFileHandler
@@ -16,13 +16,15 @@
     "DEFAULT_PROPAGATE_LOGGERS",
     "DEFAULT_QUIET_LOGGERS",
     "LogLevel",
+    "LogStream",
     "LogStyle",
     "LoggingConfig",
     "configure_logging",
 ]
 
 LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
 LogStyle = Literal["console", "json", "auto"]
+LogStream = Literal["stderr", "stdout"]
 
 DEFAULT_QUIET_LOGGERS = (
     "asyncio",
@@ -79,7 +81,7 @@ class LoggingConfig:
 
     level: LogLevel | int = "INFO"
     style: LogStyle = "auto"
-    stream: Literal["stderr", "stdout"] = "stderr"
+    stream: LogStream = "stderr"
     colors: bool | None = None
     datefmt: str = "%H:%M:%S"
     request_id_length: int = 8
@@ -158,6 +160,10 @@ def _parse_otel(raw: str) -> bool | Literal["auto"]:
 }
 
 
+# The config of the last ``configure_logging`` call, for ``_tune_loggers``.
+_applied: LoggingConfig | None = None
+
+
 def owned_handlers() -> list[logging.Handler]:
     """Return the root handlers installed by ``configure_logging``."""
     return [h for h in logging.getLogger().handlers if getattr(h, _OWNED, False)]
@@ -182,6 +188,8 @@ def _build_formatter(
 
 
 def _build_stream_handler(config: LoggingConfig) -> logging.Handler:
+    if config.stream not in get_args(LogStream):
+        raise ValueError(f"Unknown log stream {config.stream!r}")
     stream = getattr(sys, config.stream)
     is_tty = hasattr(stream, "isatty") and stream.isatty()
     colors = is_tty if config.colors is None else config.colors
@@ -236,7 +244,7 @@ def configure_logging(
         The root logger.
 
     Raises:
-        ValueError: If ``style`` is unknown.
+        ValueError: If ``style`` or ``stream`` is unknown.
         ImportError: If ``otel`` is ``True`` and the ``otel`` extra is missing.
 
     Example:
@@ -270,6 +278,17 @@ def configure_logging(
         root.addHandler(new)
     root.setLevel(config.level)
 
+    global _applied
+    _applied = config
+    _tune_loggers()
+    return root
+
+
+def _tune_loggers() -> None:
+    """Apply the logger options of the last ``configure_logging`` call."""
+    config = _applied
+    if config is None:
+        return
     for name in config.propagate:
         third_party = logging.getLogger(name)
         third_party.handlers.clear()
@@ -284,5 +303,3 @@ def configure_logging(
         logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
     for name, level in config.levels.items():
         logging.getLogger(name).setLevel(level)
-
-    return root

src/fastapi_toolsets/logger/formatters.py

@@ -12,10 +12,11 @@
 __all__ = ["ConsoleFormatter", "JsonFormatter"]
 
 # Attributes every LogRecord carries. Anything else was added through
-# ``extra`` or the context filter and is rendered as a field.
+# ``extra`` or the context filter and is rendered as a field. Uvicorn passes
+# a colored copy of its message as ``color_message``.
 _RESERVED_ATTRS = frozenset(
     logging.LogRecord("", 0, "", 0, "", None, None).__dict__
-) | {"message", "asctime"}
+) | {"message", "asctime", "color_message"}
 
 _LEVEL_COLORS = {
     "DEBUG": "\033[36m",

src/fastapi_toolsets/logger/middleware.py

@@ -33,6 +33,11 @@ class RequestLoggingMiddleware:
     The access line is logged at ``ERROR`` for a 5xx status or an unhandled
     exception, whose traceback it then carries, and at ``INFO`` otherwise.
 
+    When the app starts, the logger options of the last
+    [`configure_logging`][fastapi_toolsets.logger.configure_logging] call are
+    applied again, undoing a logging config the server loaded after importing
+    the app, as ``fastapi run`` does.
+
     Args:
         app: The ASGI application to wrap.
         request_id_header: Response header carrying the request id. ``None``
@@ -73,6 +78,11 @@ def _request_id(self, scope: Scope) -> str:
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] != "http" or (self.exclude and self.exclude(scope)):
+            if scope["type"] == "lifespan":
+                # Imported here: the config module imports this one through formatters.
+                from .config import _tune_loggers
+
+                _tune_loggers()
             await self.app(scope, receive, send)
             return
 

tests/test_logger.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import logging
+import logging.config
 import multiprocessing
 import re
 import sys
@@ -39,6 +40,7 @@
     init_logging,
     log_context,
 )
+from fastapi_toolsets.logger import config as config_module
 from fastapi_toolsets.logger import context as context_module
 from fastapi_toolsets.logger.config import DEFAULT_QUIET_LOGGERS, owned_handlers
 from fastapi_toolsets.logger.context import request_log_context
@@ -66,6 +68,7 @@ def _reset_logging():
         root.removeHandler(handler)
         handler.close()
     root.setLevel(logging.WARNING)
+    config_module._applied = None
     for name in _TOUCHED_LOGGERS:
         logging.getLogger(name).setLevel(logging.NOTSET)
 
@@ -74,6 +77,31 @@ def _reset_logging():
 _JSON = LoggingConfig(otel=False, stream="stdout", style="json")
 
 
+# The parts of uvicorn's default ``LOGGING_CONFIG`` that touch its loggers.
+_UVICORN_LOG_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "handlers": {
+        "default": {"class": "logging.StreamHandler", "stream": "ext://sys.stderr"},
+        "access": {"class": "logging.StreamHandler", "stream": "ext://sys.stdout"},
+    },
+    "loggers": {
+        "uvicorn": {"handlers": ["default"], "level": "INFO", "propagate": False},
+        "uvicorn.error": {"level": "INFO"},
+        "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
+    },
+}
+
+
+async def _run_lifespan(app: FastAPI) -> None:
+    """Start and stop *app* through its middleware, as a server does."""
+    received: asyncio.Queue[Message] = asyncio.Queue()
+    received.put_nowait({"type": "lifespan.startup"})
+    received.put_nowait({"type": "lifespan.shutdown"})
+    scope = {"type": "lifespan", "asgi": {"version": "3.0"}, "state": {}}
+    await app(scope, received.get, asyncio.Queue().put)
+
+
 def _json_lines(text: str) -> list[dict]:
     """One parsed JSON record per non-empty line of *text*."""
     return [json.loads(line) for line in text.splitlines() if line]
@@ -263,9 +291,17 @@ def test_formats_records_in_the_chosen_style(
         (line,) = capsys.readouterr().out.splitlines()
         assert re.fullmatch(pattern, line), line
 
-    def test_an_unknown_style_is_rejected(self):
-        with pytest.raises(ValueError, match="Unknown log style"):
-            configure_logging(otel=False, style="xml")
+    @pytest.mark.parametrize(
+        ("kwargs", "message"),
+        [
+            ({"style": "xml"}, "Unknown log style 'xml'"),
+            ({"stream": "stdin"}, "Unknown log stream 'stdin'"),
+        ],
+        ids=["style", "stream"],
+    )
+    def test_an_unknown_style_or_stream_is_rejected(self, kwargs, message):
+        with pytest.raises(ValueError, match=message):
+            configure_logging(otel=False, **kwargs)
 
     @pytest.mark.parametrize(
         ("kwargs", "taken_over"),
@@ -549,6 +585,12 @@ def test_json_base_keys_win_over_extra_and_unknown_types_become_strings(self):
         assert payload["timestamp"].endswith("Z")
         assert "service" not in payload
 
+    def test_uvicorn_color_message_is_rendered_by_neither_formatter(self):
+        record = _record(color_message="\033[36mhello\033[0m")
+
+        assert "color_message" not in json.loads(JsonFormatter().format(record))
+        assert "color_message" not in ConsoleFormatter().format(record)
+
     def test_exception_and_stack_are_rendered_by_both_formatters(self):
         try:
             raise RuntimeError("boom")
@@ -1089,6 +1131,42 @@ async def test_access_log_replaces_uvicorns(
         assert len(_access_lines(capsys.readouterr().out)) == access_lines_logged
         assert logging.getLogger("uvicorn.access").level == uvicorn_access_level
 
+    @pytest.mark.parametrize("setup", ["init_logging", "manual"])
+    @pytest.mark.anyio
+    async def test_startup_undoes_the_log_config_uvicorn_loads_after_the_app(
+        self, capsys, setup
+    ):
+        # ``fastapi run`` imports the app before uvicorn loads its config.
+        if setup == "init_logging":
+            app = _app()
+        else:
+            app = FastAPI()
+            configure_logging(_JSON, access_log=True)
+            app.add_middleware(RequestLoggingMiddleware)
+        logging.config.dictConfig(_UVICORN_LOG_CONFIG)
+
+        await _run_lifespan(app)
+        logging.getLogger("uvicorn.error").info("started")
+        logging.getLogger("uvicorn.access").info("GET / 200")
+        await _get(app, "/openapi.json")
+
+        out, err = capsys.readouterr()
+        lines = _json_lines(out)
+        assert err == ""
+        assert lines[0]["message"] == "started"
+        assert len(_access_lines(out)) == 1
+        assert all(line["logger"] != "uvicorn.access" for line in lines)
+
+    @pytest.mark.anyio
+    async def test_startup_keeps_a_root_level_set_after_init_logging(self):
+        app = _app()
+        # As ``caplog.set_level`` does in a test.
+        logging.getLogger().setLevel(logging.DEBUG)
+
+        await _run_lifespan(app)
+
+        assert logging.getLogger().level == logging.DEBUG
+
     @pytest.mark.anyio
     async def test_non_http_scopes_pass_through_untouched(self):
         seen = []