Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/utils/loggingutils.py: 94%

1""" 

2Helper to deal logging 

3""" 

4import logging 

5 

6_CACHE = set() 

7 

8 

9def log_once(logger: logging.Logger, level: int, msg: str, *args, **kwargs): 

10 """Helper to log a single time. 

11 

12 This is useful to log warnings which are potentially often be triggered. 

13 

14 Arguments: 

15 logger: The logger to log in 

16 level: The level of the log (i.e. `logging.WARNING`) 

17 msg: Message template 

18 args: Arguments for the message 

19 kwargs: Extra arguments to pass to the logging 

20 """ 

21 global _CACHE 

22 if not logger.isEnabledFor(level): 

23 # Is ignored so with not be displayed anyway 

24 return 

25 

26 # Resolve the message to cache the string without object ref 

27 msg = msg % args 

28 if msg in _CACHE: 

29 # Was already displayed 

30 return 

31 

32 _CACHE.add(msg) 

33 logger.log(level, msg, **kwargs) 

34 

35 

36class LoggingListener(logging.Handler): 

37 """Context listening a dedicated logger""" 

38 

39 def __init__(self, logger: logging.Logger = None): 

40 if logger is None: 

41 logger = logging.getLogger() 

42 elif not isinstance(logger, logging.Logger): 

43 logger = logging.getLogger(logger) 

44 self.logger = logger 

45 

46 self.records = [] 

47 

48 super(LoggingListener, self).__init__() 

49 

50 def __enter__(self): 

51 """Setup the state of the logger""" 

52 self.records = [] # Reset recorded LogRecords 

53 self.logger.addHandler(self) 

54 self.logger.propagate = False 

55 # ensure no log message is ignored 

56 self.entry_level = self.logger.level * 1 

57 self.logger.setLevel(logging.DEBUG) 

58 self.entry_disabled = self.logger.disabled 

59 self.logger.disabled = False 

60 return self 

61 

62 def __exit__(self, exc_type, exc_value, traceback): 

63 """Restore the previous state of the logger""" 

64 self.logger.removeHandler(self) 

65 self.logger.propagate = True 

66 self.logger.setLevel(self.entry_level) 

67 self.logger.disabled = self.entry_disabled 

68 

69 def emit(self, record): 

70 """Override `logging.Handler.emit`""" 

71 self.records.append(record)