1import logging
 2import sys
 3from typing import TextIO
 4
 5from .filters import DebugInfoFilter, WarningErrorCriticalFilter
 6from .formatters import JSONFormatter, KeyValueFormatter
 7
 8
 9def attach_log_handlers(
10    *,
11    logger: logging.Logger,
12    info_stream: TextIO,
13    warning_stream: TextIO,
14    formatter: logging.Formatter,
15) -> None:
16    """Attach two handlers to a logger that split by log level.
17
18    INFO and below go to info_stream, WARNING and above go to warning_stream.
19    """
20    # DEBUG and INFO handler
21    info_handler = logging.StreamHandler(info_stream)
22    info_handler.addFilter(DebugInfoFilter())
23    info_handler.setFormatter(formatter)
24    logger.addHandler(info_handler)
25
26    # WARNING, ERROR, and CRITICAL handler
27    warning_handler = logging.StreamHandler(warning_stream)
28    warning_handler.addFilter(WarningErrorCriticalFilter())
29    warning_handler.setFormatter(formatter)
30    logger.addHandler(warning_handler)
31
32
33def create_log_formatter(log_format: str) -> logging.Formatter:
34    """Create a formatter based on the log format setting."""
35    match log_format:
36        case "json":
37            return JSONFormatter("%(json)s")
38        case "keyvalue":
39            return KeyValueFormatter("[%(levelname)s] %(message)s %(keyvalue)s")
40        case _:
41            raise ValueError(
42                f"Invalid LOG_FORMAT: {log_format!r}. Must be 'keyvalue' or 'json'."
43            )
44
45
46def configure_logging(
47    *,
48    plain_log_level: int | str,
49    app_log_level: int | str,
50    app_log_format: str,
51    log_stream: str = "split",
52) -> None:
53    # Determine which streams to use based on log_stream setting
54    if log_stream == "split":
55        info_stream = sys.stdout
56        warning_stream = sys.stderr
57    elif log_stream == "stdout":
58        info_stream = sys.stdout
59        warning_stream = sys.stdout
60    else:  # stderr (or any other value defaults to stderr for backwards compat)
61        info_stream = sys.stderr
62        warning_stream = sys.stderr
63
64    # Determine formatter based on app_log_format
65    formatter = create_log_formatter(app_log_format)
66
67    # Create and configure the plain logger with structured formatting.
68    from .logger import app_logger
69
70    plain_logger = logging.getLogger("plain")
71    plain_logger.setLevel(plain_log_level)
72    attach_log_handlers(
73        logger=plain_logger,
74        info_stream=info_stream,
75        warning_stream=warning_stream,
76        formatter=formatter,
77    )
78    plain_logger.propagate = False
79
80    # Configure the existing app_logger
81    app_logger.setLevel(app_log_level)
82    app_logger.propagate = False
83
84    attach_log_handlers(
85        logger=app_logger,
86        info_stream=info_stream,
87        warning_stream=warning_stream,
88        formatter=formatter,
89    )
90
91    # Register the app_logger in the logging system so getLogger("app") returns it
92    logging.root.manager.loggerDict["app"] = app_logger