Coverage for stlog/base.py: 87%
141 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-06-26 07:52 +0000
1from __future__ import annotations
3import inspect
4import json
5import numbers
6import os
7import re
8import string
9import types
10from dataclasses import dataclass, field
11from string import Template
12from typing import Any, Callable, Match
14STLOG_EXTRA_KEY = "_stlog_extra"
15RICH_AVAILABLE = False
16try:
17 from rich.traceback import Traceback
19 RICH_AVAILABLE = True
20except ImportError:
21 pass
24TRUE_VALUES = ("1", "true", "yes")
25FALSE_VALUES = ("0", "false", "no")
26ALLOWED_CHARS_WITHOUT_LOGFMT_QUOTING: set = set(
27 string.ascii_letters + string.digits + ",-.@_~:"
28)
29RICH_DUMP_EXCEPTION_ON_CONSOLE_SHOW_LOCALS = (
30 os.environ.get("RICH_DUMP_EXCEPTION_ON_CONSOLE_SHOW_LOCALS", "0").lower()
31 in TRUE_VALUES
32)
35class StlogError(Exception):
36 pass
39@dataclass
40class GlobalLoggingConfig:
41 setup: bool = False
42 program_name: str = field(
43 default_factory=lambda: os.path.basename(inspect.stack()[-1][1])
44 )
45 reinject_context_in_standard_logging: bool | None = None
46 read_extra_kwargs_from_standard_logging: bool | None = None
47 _unit_tests_mode: bool = (
48 os.environ.get("STLOG_UNIT_TESTS_MODE", "0").lower() in TRUE_VALUES
49 )
52GLOBAL_LOGGING_CONFIG = GlobalLoggingConfig()
54# skip natural LogRecord attributes
55# http://docs.python.org/library/logging.html#logrecord-attributes
56# Stolen from https://github.com/madzak/python-json-logger/blob/master/src/pythonjsonlogger/jsonlogger.py
57RESERVED_ATTRS: tuple[str, ...] = (
58 "args",
59 "asctime",
60 "created",
61 "exc_info",
62 "exc_text",
63 "filename",
64 "funcName",
65 "levelname",
66 "levelno",
67 "lineno",
68 "module",
69 "msecs",
70 "message",
71 "msg",
72 "name",
73 "pathname",
74 "process",
75 "processName",
76 "relativeCreated",
77 "stack_info",
78 "thread",
79 "threadName",
80 "extra", # specific to stlog
81 "extras", # specific to stlog
82 STLOG_EXTRA_KEY, # specific to stlog
83 "rich_escaped_message", # specific to stlog
84 "rich_escaped_extras", # specific to stlog
85 "rich_level_style", # specific to stlog
86)
89def check_json_types_or_raise(to_check: Any) -> None:
90 if to_check is None:
91 return
92 if not isinstance(to_check, (dict, tuple, list, bool, str, int, float, bool)):
93 raise StlogError(
94 f"to_check should be a dict/tuple/list/bool/str/int/float/bool/None, found {type(to_check)}"
95 )
96 if isinstance(to_check, (list, tuple)):
97 for item in to_check:
98 check_json_types_or_raise(item)
99 elif isinstance(to_check, dict):
100 for key, value in to_check.items():
101 if not isinstance(key, str):
102 raise StlogError(f"dict keys should be str, found {type(key)}")
103 check_json_types_or_raise(value)
106# Adapted from https://github.com/jteppinette/python-logfmter/blob/main/logfmter/formatter.py
107def logfmt_format_string(value: str) -> str:
108 needs_dquote_escaping = '"' in value
109 needs_newline_escaping = "\n" in value
110 needs_quoting = not set(value).issubset(ALLOWED_CHARS_WITHOUT_LOGFMT_QUOTING)
111 if needs_dquote_escaping:
112 value = value.replace('"', '\\"')
113 if needs_newline_escaping:
114 value = value.replace("\n", "\\n")
115 if needs_quoting:
116 value = f'"{value}"'
117 return value if value else '""'
120# Adapted from https://github.com/jteppinette/python-logfmter/blob/main/logfmter/formatter.py
121def logfmt_format_value(value: Any) -> str:
122 if value is None:
123 return ""
124 elif isinstance(value, bool):
125 return "true" if value else "false"
126 elif isinstance(value, numbers.Number):
127 return str(value)
128 return logfmt_format_string(str(value))
131def _get_env_json_context() -> dict[str, Any]:
132 env_key = "STLOG_ENV_JSON_CONTEXT"
133 env_context = os.environ.get(env_key, None)
134 if env_context is not None:
135 try:
136 return json.loads(env_context)
137 except Exception:
138 print(
139 f"WARNING: can't load {env_key} env var value as valid JSON => ignoring"
140 )
141 return {}
144def _get_env_context() -> dict[str, Any]:
145 prefix = "STLOG_ENV_CONTEXT_"
146 res: dict[str, Any] = {}
147 for env_key in os.environ.keys():
148 if not env_key.startswith(prefix):
149 continue
150 key = env_key[len(prefix) :].lower()
151 if not key:
152 continue
153 res[key] = os.environ[env_key]
154 return res
157def get_env_context() -> dict[str, Any]:
158 if check_env_true("STLOG_IGNORE_ENV_CONTEXT", False):
159 return {}
160 env_context = {**_get_env_context(), **_get_env_json_context()}
161 for key in env_context.keys():
162 if key in RESERVED_ATTRS:
163 raise StlogError("key: %s is not allowed (reserved key)", key)
164 if not isinstance(key, str):
165 raise StlogError("key: %s must be str", key)
166 if not key.isidentifier():
167 raise StlogError(
168 "key: %s not allowed (must be a valid python identifier)", key
169 )
170 return env_context
173def check_true(value: str | None, default: bool = False) -> bool:
174 if value is None:
175 return default
176 return value.lower() in TRUE_VALUES
179def check_false(value: str | None, default: bool = False) -> bool:
180 if value is None:
181 return default
182 return value.lower() in FALSE_VALUES
185def check_env_true(env_var: str, default: bool = False) -> bool:
186 return check_true(os.environ.get(env_var, None), default)
189def check_env_false(env_var: str, default: bool = False) -> bool:
190 return check_false(os.environ.get(env_var, None), default)
193def rich_dump_exception_on_console(
194 console: Any,
195 exc_type: type[BaseException],
196 value: BaseException,
197 tb: types.TracebackType | None,
198) -> None:
199 console.print(
200 Traceback.from_exception(
201 exc_type,
202 value,
203 tb,
204 width=100,
205 extra_lines=3,
206 theme=None,
207 word_wrap=False,
208 show_locals=RICH_DUMP_EXCEPTION_ON_CONSOLE_SHOW_LOCALS,
209 locals_max_length=10,
210 locals_max_string=80,
211 locals_hide_dunder=True,
212 locals_hide_sunder=False,
213 indent_guides=True,
214 suppress=(),
215 max_frames=100,
216 )
217 )
220_ReStringMatch = Match[str] # regex match object
221_ReSubCallable = Callable[[_ReStringMatch], str] # Callable invoked by re.sub
222_EscapeSubMethod = Callable[[_ReSubCallable, str], str] # Sub method of a compiled re
225# Stolen from https://github.com/Textualize/rich/blob/master/rich/markup.py
226def rich_markup_escape(
227 markup: str,
228 _escape: _EscapeSubMethod = re.compile(r"(\\*)(\[[a-z#/@][^[]*?])").sub,
229) -> str:
230 def escape_backslashes(match: Match[str]) -> str:
231 """Called by re.sub replace matches."""
232 backslashes, text = match.groups()
233 return f"{backslashes}{backslashes}\\{text}"
235 markup = _escape(escape_backslashes, markup)
236 return markup
239# Adapted from https://github.com/madzak/python-json-logger/blob/master/src/pythonjsonlogger/jsonlogger.py
240def parse_format(fmt: str | None, style: str) -> list[str]:
241 """
242 Parses format string looking for substitutions
244 This method is responsible for returning a list of fields (as strings)
245 to include in all log messages.
246 """
247 if not fmt:
248 return []
249 if style == "$":
250 formatter_style_pattern = re.compile(r"\$\{(.+?)\}", re.IGNORECASE)
251 elif style == "{":
252 formatter_style_pattern = re.compile(r"\{(.+?)\}", re.IGNORECASE)
253 elif style == "%":
254 formatter_style_pattern = re.compile(r"%\((.+?)\)", re.IGNORECASE)
255 else:
256 raise ValueError(f"Unsupported style: {style}")
257 return formatter_style_pattern.findall(fmt)
260def format_string(fmt: str | None, style: str, record_dict: dict[str, Any]) -> str:
261 if not fmt:
262 return ""
263 if style == "$":
264 return Template(fmt).substitute(**record_dict)
265 elif style == "{":
266 return fmt.format(**record_dict)
267 elif style == "%":
268 return fmt % record_dict
269 raise StlogError(f"Invalid style: {style}")