Coverage for python/lsst/daf/butler/logging.py: 35%
229 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-11 01:27 -0700
1# This file is part of daf_butler.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28from __future__ import annotations
30__all__ = ("ButlerLogRecord", "ButlerLogRecordHandler", "ButlerLogRecords", "ButlerMDC", "JsonLogFormatter")
32import datetime
33import logging
34import traceback
35from collections.abc import Callable, Generator, Iterable, Iterator
36from contextlib import contextmanager
37from logging import Formatter, LogRecord, StreamHandler
38from types import TracebackType
39from typing import IO, Any, ClassVar, Literal, overload
41from pydantic import BaseModel, ConfigDict, PrivateAttr, RootModel
43from lsst.utils.introspection import get_full_type_name
44from lsst.utils.iteration import isplit
46_LONG_LOG_FORMAT = "{levelname} {asctime} {name} {filename}:{lineno} - {message}"
47"""Default format for log records."""
50class MDCDict(dict):
51 """Dictionary for MDC data.
53 This is internal class used for better formatting of MDC in Python logging
54 output. It behaves like `defaultdict(str)` but overrides ``__str__`` and
55 ``__repr__`` method to produce output better suited for logging records.
56 """
58 def __getitem__(self, name: str) -> str:
59 """Return value for a given key or empty string for missing key."""
60 return self.get(name, "")
62 def __str__(self) -> str:
63 """Return string representation, strings are interpolated without
64 quotes.
65 """
66 items = (f"{k}={self[k]}" for k in sorted(self))
67 return "{" + ", ".join(items) + "}"
69 def __repr__(self) -> str:
70 return str(self)
73class ButlerMDC:
74 """Handle setting and unsetting of global MDC records.
76 The Mapped Diagnostic Context (MDC) can be used to set context
77 for log messages.
79 Currently there is one global MDC dict. Per-thread MDC is not
80 yet supported.
81 """
83 _MDC = MDCDict()
85 _old_factory: Callable[..., logging.LogRecord] | None = None
86 """Old log record factory."""
88 @classmethod
89 def MDC(cls, key: str, value: str) -> str:
90 """Set MDC for this key to the supplied value.
92 Parameters
93 ----------
94 key : `str`
95 Key to modify.
96 value : `str`
97 New value to use.
99 Returns
100 -------
101 old : `str`
102 The previous value for this key.
103 """
104 old_value = cls._MDC[key]
105 cls._MDC[key] = value
106 return old_value
108 @classmethod
109 def MDCRemove(cls, key: str) -> None:
110 """Clear the MDC value associated with this key.
112 Can be called even if the key is not known to MDC.
114 Parameters
115 ----------
116 key : `str`
117 Key for which the MDC value should be removed.
118 """
119 cls._MDC.pop(key, None)
121 @classmethod
122 def clear_mdc(cls) -> None:
123 """Clear all MDC entries."""
124 cls._MDC.clear()
126 @classmethod
127 @contextmanager
128 def set_mdc(cls, mdc: dict[str, str]) -> Generator[None, None, None]:
129 """Set the MDC key for this context.
131 This context manager also adds a mapping named ``mdc`` to any
132 exceptions that escape it.
134 Parameters
135 ----------
136 mdc : `dict` of `str`, `str`
137 MDC keys to update temporarily.
139 Notes
140 -----
141 Other MDC keys are not modified. The previous values are restored
142 on exit (removing them if the were unset previously).
143 """
144 previous = {}
145 for k, v in mdc.items():
146 previous[k] = cls.MDC(k, v)
148 try:
149 yield
150 except BaseException as e:
151 # In logging, inner context overrules outer context. Need the same
152 # for exceptions.
153 inner_context: MDCDict = e.mdc if hasattr(e, "mdc") else MDCDict()
154 e.mdc = cls._MDC.copy() | inner_context # type: ignore[attr-defined]
155 raise
156 finally:
157 for k, v in previous.items():
158 if not v:
159 cls.MDCRemove(k)
160 else:
161 cls.MDC(k, v)
163 @classmethod
164 def add_mdc_log_record_factory(cls) -> None:
165 """Add a log record factory that adds a MDC record to `LogRecord`."""
166 old_factory = logging.getLogRecordFactory()
168 # Need explicit args in case exc_info is called as an arg (and because
169 # the factory API has a parameter literally called `args`).
170 def record_factory(
171 name: str,
172 level: int,
173 fn: str,
174 lno: int,
175 msg: str,
176 args: tuple,
177 exc_info: tuple | None | Literal[False],
178 func: str | None = None,
179 sinfo: TracebackType | None = None,
180 **kwargs: Any,
181 ) -> LogRecord:
182 record = old_factory(name, level, fn, lno, msg, args, exc_info, func, sinfo, **kwargs)
183 # Make sure we send a copy of the global dict in the record.
184 mdc = MDCDict(cls._MDC)
185 if exc_info is not None and exc_info is not False:
186 _, ex, _ = exc_info
187 # TODO: this doesn't handle chained exceptions, fix on DM-47546
188 if hasattr(ex, "mdc"):
189 # Context at the point where the exception was raised
190 # takes precedence.
191 mdc.update(ex.mdc)
192 record.MDC = mdc
193 return record
195 cls._old_factory = old_factory
196 logging.setLogRecordFactory(record_factory)
198 @classmethod
199 def restore_log_record_factory(cls) -> None:
200 """Restores the log record factory to the original form.
202 Does nothing if there has not been a call to
203 `add_mdc_log_record_factory`.
204 """
205 if cls._old_factory:
206 logging.setLogRecordFactory(cls._old_factory)
209class ButlerLogRecord(BaseModel):
210 """A model representing a `logging.LogRecord`.
212 A `~logging.LogRecord` always uses the current time in its record
213 when recreated and that makes it impossible to use it as a
214 serialization format. Instead have a local representation of a
215 `~logging.LogRecord` that matches Butler needs.
216 """
218 _log_format: ClassVar[str] = _LONG_LOG_FORMAT
220 name: str
221 asctime: datetime.datetime
222 message: str
223 levelno: int
224 levelname: str
225 filename: str
226 pathname: str
227 lineno: int
228 funcName: str | None = None
229 process: int
230 processName: str
231 exc_info: str | None = None
232 MDC: dict[str, str]
234 model_config = ConfigDict(frozen=True)
236 @classmethod
237 def from_record(cls, record: LogRecord) -> ButlerLogRecord:
238 """Create a new instance from a `~logging.LogRecord`.
240 Parameters
241 ----------
242 record : `logging.LogRecord`
243 The record from which to extract the relevant information.
244 """
245 # The properties that are one-to-one mapping.
246 simple = (
247 "name",
248 "levelno",
249 "levelname",
250 "filename",
251 "pathname",
252 "lineno",
253 "funcName",
254 "process",
255 "processName",
256 )
258 record_dict = {k: getattr(record, k) for k in simple}
260 record_dict["message"] = record.getMessage()
262 # MDC -- ensure the contents are copied to prevent any confusion
263 # over the MDC global being updated later.
264 record_dict["MDC"] = dict(getattr(record, "MDC", {}))
266 # Always use UTC because in distributed systems we can't be sure
267 # what timezone localtime is and it's easier to compare logs if
268 # every system is using the same time.
269 record_dict["asctime"] = datetime.datetime.fromtimestamp(record.created, tz=datetime.UTC)
271 # Sometimes exception information is included so must be
272 # extracted.
273 if record.exc_info:
274 etype = record.exc_info[0]
275 evalue = record.exc_info[1]
276 tb = record.exc_info[2]
277 record_dict["exc_info"] = "\n".join(traceback.format_exception(etype, evalue, tb))
279 return cls(**record_dict)
281 def format(self, log_format: str | None = None) -> str:
282 """Format this record.
284 Parameters
285 ----------
286 log_format : `str`, optional
287 The format string to use. This string follows the standard
288 f-style use for formatting log messages. If `None`
289 the class default will be used.
291 Returns
292 -------
293 text : `str`
294 The formatted log message.
295 """
296 if log_format is None:
297 log_format = self._log_format
299 as_dict = self.model_dump()
301 # Special case MDC content. Convert it to an MDCDict
302 # so that missing items do not break formatting.
303 as_dict["MDC"] = MDCDict(as_dict["MDC"])
305 as_dict["asctime"] = as_dict["asctime"].isoformat()
306 formatted = log_format.format(**as_dict)
307 return formatted
309 def __str__(self) -> str:
310 return self.format()
313# The class below can convert LogRecord to ButlerLogRecord if needed.
314Record = LogRecord | ButlerLogRecord
317class _ButlerLogRecords(RootModel):
318 root: list[ButlerLogRecord]
321# Do not inherit from MutableSequence since mypy insists on the values
322# being Any even though we wish to constrain them to Record.
323class ButlerLogRecords(_ButlerLogRecords):
324 """Class representing a collection of `ButlerLogRecord`."""
326 _log_format: str | None = PrivateAttr(None)
328 @classmethod
329 def from_records(cls, records: Iterable[ButlerLogRecord]) -> ButlerLogRecords:
330 """Create collection from iterable.
332 Parameters
333 ----------
334 records : iterable of `ButlerLogRecord`
335 The records to seed this class with.
336 """
337 return cls.model_construct(root=list(records))
339 @classmethod
340 def from_file(cls, filename: str) -> ButlerLogRecords:
341 """Read records from file.
343 Parameters
344 ----------
345 filename : `str`
346 Name of file containing the JSON records.
348 Notes
349 -----
350 Works with one-record-per-line format JSON files and a direct
351 serialization of the Pydantic model.
352 """
353 with open(filename) as fd:
354 return cls.from_stream(fd)
356 @staticmethod
357 def _detect_model(startdata: str | bytes) -> bool:
358 """Given some representative data, determine if this is a serialized
359 model or a streaming format.
361 Parameters
362 ----------
363 startdata : `bytes` or `str`
364 Representative characters or bytes from the start of a serialized
365 collection of log records.
367 Returns
368 -------
369 is_model : `bool`
370 Returns `True` if the data look like a serialized pydantic model.
371 Returns `False` if it looks like a streaming format. Returns
372 `False` also if an empty string is encountered since this
373 is not understood by `ButlerLogRecords.model_validate_json()`.
375 Raises
376 ------
377 ValueError
378 Raised if the sentinel doesn't look like either of the supported
379 log record formats.
380 """
381 if not startdata:
382 return False
384 # Allow byte or str streams since pydantic supports either.
385 # We don't want to convert the entire input to unicode unnecessarily.
386 error_type = "str"
387 if isinstance(startdata, bytes):
388 first_char = chr(startdata[0])
389 error_type = "byte"
390 else:
391 first_char = startdata[0]
393 if first_char == "[":
394 # This is an array of records.
395 return True
396 if first_char != "{":
397 # Limit the length of string reported in error message in case
398 # this is an enormous file.
399 max = 32
400 if len(startdata) > max:
401 startdata = f"{startdata[:max]!r}..."
402 raise ValueError(
403 "Unrecognized JSON log format. Expected '{' or '[' but got"
404 f" {first_char!r} from {error_type} content starting with {startdata!r}"
405 )
407 # Assume a record per line.
408 return False
410 @classmethod
411 def from_stream(cls, stream: IO) -> ButlerLogRecords:
412 """Read records from I/O stream.
414 Parameters
415 ----------
416 stream : `typing.IO`
417 Stream from which to read JSON records.
419 Notes
420 -----
421 Works with one-record-per-line format JSON files and a direct
422 serialization of the Pydantic model.
423 """
424 first_line = stream.readline()
426 if not first_line:
427 # Empty file, return zero records.
428 return cls.from_records([])
430 is_model = cls._detect_model(first_line)
432 if is_model:
433 # This is a ButlerLogRecords model serialization so all the
434 # content must be read first.
435 all = first_line + stream.read()
436 return cls.model_validate_json(all)
438 # A stream of records with one record per line.
439 records = [ButlerLogRecord.model_validate_json(first_line)]
440 for line in stream:
441 line = line.rstrip()
442 if line: # Filter out blank lines.
443 records.append(ButlerLogRecord.model_validate_json(line))
445 return cls.from_records(records)
447 @classmethod
448 def from_raw(cls, serialized: str | bytes) -> ButlerLogRecords:
449 """Parse raw serialized form and return records.
451 Parameters
452 ----------
453 serialized : `bytes` or `str`
454 Either the serialized JSON of the model created using
455 ``.model_dump_json()`` or a streaming format of one JSON
456 `ButlerLogRecord` per line. This can also support a zero-length
457 string.
458 """
459 if not serialized:
460 # No records to return
461 return cls.from_records([])
463 # Only send the first character for analysis.
464 is_model = cls._detect_model(serialized)
466 if is_model:
467 return cls.model_validate_json(serialized)
469 # Filter out blank lines -- mypy is confused by the newline
470 # argument to isplit() [which can't have two different types
471 # simultaneously] so we have to duplicate some logic.
472 substrings: Iterator[str | bytes]
473 if isinstance(serialized, str):
474 substrings = isplit(serialized, "\n")
475 elif isinstance(serialized, bytes):
476 substrings = isplit(serialized, b"\n")
477 else:
478 raise TypeError(f"Serialized form must be str or bytes not {get_full_type_name(serialized)}")
479 records = [ButlerLogRecord.model_validate_json(line) for line in substrings if line]
481 return cls.from_records(records)
483 @property
484 def log_format(self) -> str:
485 if self._log_format is None:
486 return _LONG_LOG_FORMAT
487 return self._log_format
489 # Pydantic does not allow a property setter to be given for
490 # public properties of a model that is not based on a dict.
491 def set_log_format(self, format: str | None) -> str | None:
492 """Set the log format string for these records.
494 Parameters
495 ----------
496 format : `str`, optional
497 The new format string to use for converting this collection
498 of records into a string. If `None` the default format will be
499 used.
501 Returns
502 -------
503 old_format : `str`, optional
504 The previous log format.
505 """
506 previous = self._log_format
507 self._log_format = format
508 return previous
510 def __len__(self) -> int:
511 return len(self.root)
513 # The signature does not match the one in BaseModel but that is okay
514 # if __root__ is being used.
515 # See https://pydantic-docs.helpmanual.io/usage/models/#custom-root-types
516 def __iter__(self) -> Iterator[ButlerLogRecord]: # type: ignore
517 return iter(self.root)
519 def __setitem__(self, index: int, value: Record) -> None:
520 self.root[index] = self._validate_record(value)
522 @overload
523 def __getitem__(self, index: int) -> ButlerLogRecord: ... 523 ↛ exitline 523 didn't return from function '__getitem__' because
525 @overload
526 def __getitem__(self, index: slice) -> ButlerLogRecords: ... 526 ↛ exitline 526 didn't return from function '__getitem__' because
528 def __getitem__(self, index: slice | int) -> ButlerLogRecords | ButlerLogRecord:
529 # Handles slices and returns a new collection in that
530 # case.
531 item = self.root[index]
532 if isinstance(item, list):
533 return type(self)(item)
534 else:
535 return item
537 def __reversed__(self) -> Iterator[ButlerLogRecord]:
538 return self.root.__reversed__()
540 def __delitem__(self, index: slice | int) -> None:
541 del self.root[index]
543 def __str__(self) -> str:
544 # Ensure that every record uses the same format string.
545 return "\n".join(record.format(self.log_format) for record in self.root)
547 def _validate_record(self, record: Record) -> ButlerLogRecord:
548 if isinstance(record, ButlerLogRecord):
549 pass
550 elif isinstance(record, LogRecord):
551 record = ButlerLogRecord.from_record(record)
552 else:
553 raise ValueError(f"Can only append item of type {type(record)}")
554 return record
556 def insert(self, index: int, value: Record) -> None:
557 self.root.insert(index, self._validate_record(value))
559 def append(self, value: Record) -> None:
560 value = self._validate_record(value)
561 self.root.append(value)
563 def clear(self) -> None:
564 self.root.clear()
566 def extend(self, records: Iterable[Record]) -> None:
567 self.root.extend(self._validate_record(record) for record in records)
569 def pop(self, index: int = -1) -> ButlerLogRecord:
570 return self.root.pop(index)
572 def reverse(self) -> None:
573 self.root.reverse()
576class ButlerLogRecordHandler(StreamHandler):
577 """Python log handler that accumulates records."""
579 def __init__(self) -> None:
580 super().__init__()
581 self.records = ButlerLogRecords.model_construct(root=[])
583 def emit(self, record: LogRecord) -> None:
584 self.records.append(record)
587class JsonLogFormatter(Formatter):
588 """Format a `LogRecord` in JSON format."""
590 def format(self, record: LogRecord) -> str:
591 butler_record = ButlerLogRecord.from_record(record)
592 return butler_record.model_dump_json(exclude_unset=True, exclude_defaults=True)