Coverage for stlog/setup.py: 75%
91 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-06-26 07:52 +0000
« prev ^ index » next coverage.py v7.8.0, created at 2025-06-26 07:52 +0000
1from __future__ import annotations
3import logging
4import os
5import sys
6import traceback
7import types
8import typing
9import warnings
11from stlog.adapter import getLogger
12from stlog.base import GLOBAL_LOGGING_CONFIG, check_env_false
13from stlog.formatter import (
14 DEFAULT_STLOG_GCP_JSON_FORMAT,
15 JsonFormatter,
16)
17from stlog.output import Output, StreamOutput, make_stream_or_rich_stream_output
19DEFAULT_LEVEL: str = os.environ.get("STLOG_LEVEL", "INFO")
20DEFAULT_CAPTURE_WARNINGS: bool = check_env_false("STLOG_CAPTURE_WARNINGS", True)
21DEFAULT_PROGRAM_NAME: str | None = os.environ.get("STLOG_PROGRAM_NAME", None)
22DEFAULT_DESTINATION: str = os.environ.get("STLOG_DESTINATION", "stderr").lower()
23DEFAULT_OUTPUT: str = os.environ.get("STLOG_OUTPUT", "console").lower()
26def _make_default_stream() -> typing.TextIO:
27 if DEFAULT_DESTINATION == "stderr":
28 return sys.stderr
29 elif DEFAULT_DESTINATION == "stdout":
30 return sys.stdout
31 raise Exception(
32 f"bad value:{DEFAULT_DESTINATION} for STLOG_DESTINATION env var => must be 'stderr' or 'stdout'"
33 )
36def _make_default_outputs() -> list[Output]:
37 if DEFAULT_OUTPUT == "console":
38 return [make_stream_or_rich_stream_output(stream=_make_default_stream())]
39 elif DEFAULT_OUTPUT == "json":
40 return [
41 StreamOutput(
42 stream=_make_default_stream(),
43 formatter=JsonFormatter(),
44 )
45 ]
46 elif DEFAULT_OUTPUT == "json-human":
47 return [
48 StreamOutput(
49 stream=_make_default_stream(),
50 formatter=JsonFormatter(indent=4),
51 )
52 ]
53 elif DEFAULT_OUTPUT == "json-gcp":
54 return [
55 StreamOutput(
56 stream=_make_default_stream(),
57 formatter=JsonFormatter(
58 fmt=DEFAULT_STLOG_GCP_JSON_FORMAT,
59 ),
60 )
61 ]
62 else:
63 raise Exception(
64 f"bad value:{DEFAULT_OUTPUT} for STLOG_OUTPUT env var => must be 'console', 'json', 'json-human' or 'json-gcp'"
65 )
68def _logging_excepthook(
69 exc_type: type[BaseException],
70 value: BaseException,
71 tb: types.TracebackType | None = None,
72) -> None:
73 if issubclass(exc_type, KeyboardInterrupt):
74 sys.__excepthook__(exc_type, value, tb)
75 return
76 try:
77 program_logger = getLogger(GLOBAL_LOGGING_CONFIG.program_name)
78 program_logger.error(
79 "Exception catched in excepthook", exc_info=(exc_type, value, tb)
80 )
81 except Exception:
82 print(
83 "ERROR: Exception during exception handling => let's dump this on standard output"
84 )
85 print(traceback.format_exc(), file=sys.stderr)
88def setup( # noqa: PLR0913
89 *,
90 level: str | int = DEFAULT_LEVEL,
91 outputs: typing.Iterable[Output] | None = None,
92 program_name: str | None = DEFAULT_PROGRAM_NAME,
93 capture_warnings: bool = DEFAULT_CAPTURE_WARNINGS,
94 logging_excepthook: typing.Callable[
95 [type[BaseException], BaseException, types.TracebackType | None],
96 typing.Any,
97 ]
98 | None = _logging_excepthook,
99 extra_levels: typing.Mapping[str, str | int] = {},
100 reinject_context_in_standard_logging: bool | None = None,
101 read_extra_kwargs_from_standard_logging: bool | None = None,
102) -> None:
103 """Set up the Python logging with stlog (globally).
105 This removes all existing handlers and
106 sets up handlers/formatters/... for Python logging.
108 Args:
109 level: the root log level as int or as a string representation (in uppercase)
110 (see https://docs.python.org/3/library/logging.html#levels), the default
111 value is read in STLOG_LEVEL env var or set to INFO (if not set).
112 outputs: iterable of Output to log to.
113 program_name: the name of the program, the default value is read in STLOG_PROGRAM_NAME
114 env var or auto-detected if not set.
115 capture_warnings: capture warnings from the `warnings` module.
116 logging_excepthook: if not None, override sys.excepthook with the given callable
117 See https://docs.python.org/3/library/sys.html#sys.excepthook for details.
118 extra_levels: dict "logger name => log level" for quick override of
119 the root log level (for some loggers).
120 reinject_context_in_standard_logging: if True, reinject the LogContext
121 in log record emitted with python standard loggers (note: can be overriden per `stlog.output.Output`,
122 default to `STLOG_REINJECT_CONTEXT_IN_STANDARD_LOGGING` env var or True if not set).
123 read_extra_kwargs_from_standard_logging: if try to reinject the extra kwargs from standard logging
124 (note: can be overriden per `stlog.output.Output`, default to `STLOG_READ_EXTRA_KWARGS_FROM_STANDARD_LOGGING` env var
125 or False if not set).
127 """
128 GLOBAL_LOGGING_CONFIG.reinject_context_in_standard_logging = (
129 reinject_context_in_standard_logging
130 )
131 GLOBAL_LOGGING_CONFIG.read_extra_kwargs_from_standard_logging = (
132 read_extra_kwargs_from_standard_logging
133 )
134 if program_name is not None:
135 GLOBAL_LOGGING_CONFIG.program_name = program_name
137 if GLOBAL_LOGGING_CONFIG._unit_tests_mode:
138 # remove all configured loggers
139 for key in list(logging.Logger.manager.loggerDict.keys()):
140 logging.Logger.manager.loggerDict.pop(key)
142 root_logger = logging.getLogger(None)
143 # Remove all handlers
144 for handler in list(root_logger.handlers):
145 root_logger.removeHandler(handler)
147 # Add configured handlers
148 if outputs is None:
149 outputs = _make_default_outputs()
150 for out in outputs:
151 root_logger.addHandler(out.get_handler())
153 root_logger.setLevel(level)
155 if logging_excepthook:
156 sys.excepthook = logging_excepthook
158 if capture_warnings:
159 if GLOBAL_LOGGING_CONFIG._unit_tests_mode:
160 # to avoid the capture by pytest
161 logging._warnings_showwarning = None # type: ignore
162 logging.captureWarnings(True)
164 for lgger, lvel in extra_levels.items():
165 logging.getLogger(lgger).setLevel(lvel)
167 GLOBAL_LOGGING_CONFIG.setup = True
170ROOT_LOGGER = getLogger("root")
173def log(level: int, msg, *args, **kwargs):
174 """Log a message with the given integer severity level' on the root logger.
176 `setup()` is automatically called with default arguments if not already done before.
177 """
178 if not GLOBAL_LOGGING_CONFIG.setup:
179 setup()
180 ROOT_LOGGER.log(level, msg, *args, **kwargs)
183def debug(msg, *args, **kwargs):
184 """Log a message with severity 'DEBUG' on the root logger.
186 `setup()` is automatically called with default arguments if not already done before.
187 """
188 log(logging.DEBUG, msg, *args, **kwargs)
191def info(msg, *args, **kwargs):
192 """Log a message with severity 'INFO' on the root logger.
194 `setup()` is automatically called with default arguments if not already done before.
195 """
196 log(logging.INFO, msg, *args, **kwargs)
199def warning(msg, *args, **kwargs):
200 """Log a message with severity 'WARNING' on the root logger.
202 `setup()` is automatically called with default arguments if not already done before.
203 """
204 log(logging.WARNING, msg, *args, **kwargs)
207def warn(msg, *args, **kwargs):
208 if (
209 not GLOBAL_LOGGING_CONFIG.setup
210 ): # we do this here to be able to capture the next warning
211 setup()
212 warnings.warn(
213 "The 'warn' function is deprecated, use 'warning' instead",
214 DeprecationWarning,
215 2,
216 )
217 warning(msg, *args, **kwargs)
220def error(msg, *args, **kwargs):
221 """Log a message with severity 'ERROR' on the root logger.
223 `setup()` is automatically called with default arguments if not already done before.
224 """
225 log(logging.ERROR, msg, *args, **kwargs)
228def critical(msg, *args, **kwargs):
229 """Log a message with severity 'CRITICAL' on the root logger.
231 `setup()` is automatically called with default arguments if not already done before.
232 """
233 log(logging.CRITICAL, msg, *args, **kwargs)
236fatal = critical
239def exception(msg, *args, exc_info=True, **kwargs):
240 """Log a message with severity 'ERROR' on the root logger with exception information.
242 `setup()` is automatically called with default arguments if not already done before.
243 """
244 error(msg, *args, exc_info=exc_info, **kwargs)