Coverage for frappe_manager / logger / contextual.py: 94%

1""" 

2Contextual logger adapter. 

3 

4This module provides ContextualLogger which wraps a standard Python logger 

5to automatically add context information (bench name, operation, component) 

6to all log messages. 

7""" 

8 

9import logging 

10from typing import Any 

11 

12from frappe_manager.logger.context import LoggerContext 

13 

14 

15class ContextualLogger: 

16 """ 

17 Adapter that wraps logging.Logger to add context to all messages. 

18 

19 Automatically prefixes all log messages with formatted context information. 

20 Supports creating child loggers with extended context for nested operations. 

21 

22 Example: 

23 >>> import logging 

24 >>> base_logger = logging.getLogger("fm") 

25 >>> context = LoggerContext(bench="mybench", operation="create") 

26 >>> logger = ContextualLogger(base_logger, context) 

27 >>> logger.info("Starting operation") 

28 # Logs: "[bench=mybench] [op=create] Starting operation" 

29 

30 >>> child = logger.child(component="docker") 

31 >>> child.info("Building containers") 

32 # Logs: "[bench=mybench] [op=create] [component=docker] Building containers" 

33 

34 >>> logger.info("Site created", extra_fields={"environment": "dev", "apps": "frappe,erpnext"}) 

35 # Logs: "[bench=mybench] [op=create] Site created | environment=dev | apps=frappe,erpnext" 

36 """ 

37 

38 def __init__(self, logger: logging.Logger, context: LoggerContext | None = None): 

39 """ 

40 Initialize contextual logger. 

41 

42 Args: 

43 logger: Base Python logger to wrap 

44 context: Optional context to add to messages (defaults to empty context) 

45 """ 

46 self.logger = logger 

47 self.context = context or LoggerContext() 

48 

49 def _format_message(self, msg: str, extra_fields: dict[str, Any] | None = None) -> str: 

50 prefix = self.context.format() 

51 extra_suffix = ( 

52 self._format_extra_fields(extra_fields) if extra_fields and isinstance(extra_fields, dict) else "" 

53 ) 

54 

55 if prefix and extra_suffix: 

56 return f"{prefix} {msg}{extra_suffix}" 

57 if prefix: 

58 return f"{prefix} {msg}" 

59 if extra_suffix: 

60 return f"{msg}{extra_suffix}" 

61 return msg 

62 

63 def _format_extra_fields(self, extra_fields: dict[str, Any]) -> str: 

64 if not extra_fields: 

65 return "" 

66 

67 parts = [] 

68 for key, value in extra_fields.items(): 

69 if value is not None: 

70 parts.append(f"{key}={value}") 

71 

72 if parts: 

73 return " | " + " | ".join(parts) 

74 return "" 

75 

76 def debug(self, msg: str, *args, extra_fields: dict[str, Any] | None = None, **kwargs): 

77 self.logger.debug(self._format_message(msg, extra_fields), *args, **kwargs) 

78 

79 def info(self, msg: str, *args, extra_fields: dict[str, Any] | None = None, **kwargs): 

80 self.logger.info(self._format_message(msg, extra_fields), *args, **kwargs) 

81 

82 def warning(self, msg: str, *args, extra_fields: dict[str, Any] | None = None, **kwargs): 

83 self.logger.warning(self._format_message(msg, extra_fields), *args, **kwargs) 

84 

85 def error(self, msg: str, *args, extra_fields: dict[str, Any] | None = None, **kwargs): 

86 self.logger.error(self._format_message(msg, extra_fields), *args, **kwargs) 

87 

88 def exception(self, msg: str, *args, extra_fields: dict[str, Any] | None = None, **kwargs): 

89 self.logger.exception(self._format_message(msg, extra_fields), *args, **kwargs) 

90 

91 def child(self, **context_overrides) -> "ContextualLogger": 

92 """ 

93 Create child logger with extended context. 

94 

95 The child inherits all context from parent and can override or extend it. 

96 Useful for nested operations that need additional context. 

97 

98 Args: 

99 **context_overrides: Context fields to add/override (bench, operation, component, extra) 

100 

101 Returns: 

102 New ContextualLogger with extended context 

103 

104 Example: 

105 >>> parent = ContextualLogger(logger, LoggerContext(bench="mybench")) 

106 >>> child = parent.child(operation="create", component="docker") 

107 >>> child.info("Starting containers") 

108 # Logs: "[bench=mybench] [op=create] [component=docker] Starting containers" 

109 """ 

110 new_context = self.context.child(**context_overrides) 

111 return ContextualLogger(self.logger, new_context) 

112 

113 # Pass-through properties and methods for compatibility 

114 

115 @property 

116 def level(self) -> int: 

117 """Get the logger's current level.""" 

118 return self.logger.level 

119 

120 def setLevel(self, level: int): 

121 """ 

122 Set the logger's level. 

123 

124 Args: 

125 level: Logging level (logging.DEBUG, logging.INFO, etc.) 

126 """ 

127 self.logger.setLevel(level) 

128 

129 def isEnabledFor(self, level: int) -> bool: 

130 """ 

131 Check if logger is enabled for a given level. 

132 

133 Args: 

134 level: Logging level to check 

135 

136 Returns: 

137 True if logger would emit a message at this level 

138 """ 

139 return self.logger.isEnabledFor(level) 

140 

141 @property 

142 def name(self) -> str: 

143 """Get the logger's name.""" 

144 return self.logger.name 

145 

146 def get_context(self) -> LoggerContext: 

147 """ 

148 Get the current context. 

149 

150 Returns: 

151 Current LoggerContext 

152 """ 

153 return self.context