Coverage for stlog/adapter.py: 94%

1from __future__ import annotations 

2 

3import collections 

4import logging 

5import typing 

6 

7from stlog.base import ( 

8 RESERVED_ATTRS, 

9 STLOG_EXTRA_KEY, 

10 check_json_types_or_raise, 

11) 

12from stlog.context import LogContext 

13 

14 

15# ligthly adapted from https://github.com/Mergifyio/daiquiri/blob/main/daiquiri/__init__.py 

16class _KeywordArgumentAdapter(logging.LoggerAdapter): 

17 """Logger adapter to add keyword arguments to log record's extra data. 

18 

19 Keywords passed to the log call are added to the "extra" 

20 dictionary passed to the underlying logger so they are emitted 

21 with the log message and available to the format string. 

22 """ 

23 

24 def process( 

25 self, msg: typing.Any, kwargs: collections.abc.MutableMapping[str, typing.Any] 

26 ) -> tuple[typing.Any, collections.abc.MutableMapping[str, typing.Any]]: 

27 # Make a new extra dictionary combining the values we were 

28 # given when we were constructed and anything from kwargs. 

29 if self.extra is not None: 

30 # kvs passed during getLogger() call 

31 check_json_types_or_raise(self.extra) 

32 extra = dict(self.extra) 

33 if kwargs.get("extra"): 

34 # when you use the "extra" standard kwargs at log() time 

35 extra.update(kwargs.pop("extra")) 

36 # Move any unknown keyword arguments into the extra 

37 # dictionary. 

38 for name in list(kwargs.keys()): 

39 if name in RESERVED_ATTRS: 

40 continue 

41 extra[name] = kwargs.pop(name) 

42 extra[STLOG_EXTRA_KEY] = set(extra.keys()) 

43 kwargs["extra"] = extra 

44 return msg, kwargs 

45 

46 

47class StLogLoggerAdapter(_KeywordArgumentAdapter): 

48 """stlog `LoggerAdapter` with `stlog.LogContext` support.""" 

49 

50 def __init__(self, logger, extra, ignore_global_logging_context: bool = False): 

51 self.ignore_global_logging_context = ignore_global_logging_context 

52 super().__init__(logger, extra) 

53 

54 def process( 

55 self, msg: typing.Any, kwargs: collections.abc.MutableMapping[str, typing.Any] 

56 ) -> tuple[typing.Any, collections.abc.MutableMapping[str, typing.Any]]: 

57 if self.ignore_global_logging_context: 

58 new_kwargs = dict(kwargs) 

59 else: 

60 new_kwargs = {**LogContext._get(), **kwargs} 

61 return super().process(msg, new_kwargs) 

62 

63 def addFilter(self, filter): # noqa: N802 

64 self.logger.addFilter(filter) 

65 

66 def removeFilter(self, filter): # noqa: N802 

67 self.logger.removeFilter(filter) 

68 

69 

70def getLogger(name: str | None = None, **kwargs) -> StLogLoggerAdapter: # noqa: N802 

71 """Return a standard logger (adapted for `stlog` and `stlog.LogContext` support). 

72 

73 You can pass some context key/values in `**kwargs` which will be specific to this logger. 

74 

75 If you want to set more globally available context, use `stlog.LogContext` class. 

76 

77 Args: 

78 name: logger name. 

79 """ 

80 return StLogLoggerAdapter(logging.getLogger(name), kwargs)