Coverage for src/symphra_modules/utils/logger.py: 78.38%

1"""抽象日志接口.""" 

2 

3import logging 

4from typing import Any, Protocol 

5 

6 

7class Logger(Protocol): 

8 """日志记录器协议. 

9 

10 定义了日志接口,用户可以提供自己的日志实现。 

11 """ 

12 

13 def debug(self, message: str, **kwargs: Any) -> None: 

14 """记录 DEBUG 级别日志.""" 

15 ... 

16 

17 def info(self, message: str, **kwargs: Any) -> None: 

18 """记录 INFO 级别日志.""" 

19 ... 

20 

21 def warning(self, message: str, **kwargs: Any) -> None: 

22 """记录 WARNING 级别日志.""" 

23 ... 

24 

25 def error(self, message: str, **kwargs: Any) -> None: 

26 """记录 ERROR 级别日志.""" 

27 ... 

28 

29 def exception(self, message: str, **kwargs: Any) -> None: 

30 """记录异常信息.""" 

31 ... 

32 

33 

34class StandardLogger: 

35 """基于 Python 标准库 logging 的默认日志实现.""" 

36 

37 def __init__(self, name: str = "symphra_modules") -> None: 

38 """初始化日志记录器. 

39 

40 Args: 

41 name: 日志记录器名称 

42 """ 

43 self._logger = logging.getLogger(name) 

44 if not self._logger.handlers: 44 ↛ exitline 44 didn't return from function '__init__' because the condition on line 44 was always true

45 handler = logging.StreamHandler() 

46 formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") 

47 handler.setFormatter(formatter) 

48 self._logger.addHandler(handler) 

49 self._logger.setLevel(logging.INFO) 

50 

51 def debug(self, message: str, **kwargs: Any) -> None: 

52 """记录 DEBUG 级别日志.""" 

53 self._logger.debug(self._format_message(message, kwargs)) 

54 

55 def info(self, message: str, **kwargs: Any) -> None: 

56 """记录 INFO 级别日志.""" 

57 self._logger.info(self._format_message(message, kwargs)) 

58 

59 def warning(self, message: str, **kwargs: Any) -> None: 

60 """记录 WARNING 级别日志.""" 

61 self._logger.warning(self._format_message(message, kwargs)) 

62 

63 def error(self, message: str, **kwargs: Any) -> None: 

64 """记录 ERROR 级别日志.""" 

65 self._logger.error(self._format_message(message, kwargs)) 

66 

67 def exception(self, message: str, **kwargs: Any) -> None: 

68 """记录异常信息.""" 

69 self._logger.exception(self._format_message(message, kwargs)) 

70 

71 def _format_message(self, message: str, kwargs: dict[str, Any]) -> str: 

72 """格式化日志消息. 

73 

74 Args: 

75 message: 消息模板 

76 kwargs: 格式化参数 

77 

78 Returns: 

79 格式化后的消息 

80 """ 

81 if kwargs: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 try: 

83 return message.format(**kwargs) 

84 except (KeyError, ValueError): 

85 # 如果格式化失败，附加 kwargs 信息 

86 return f"{message} | {kwargs}" 

87 return message 

88 

89 

90# 全局日志实例 

91_global_logger: Logger = StandardLogger() 

92 

93 

94def get_logger() -> Logger: 

95 """获取全局日志实例. 

96 

97 Returns: 

98 当前配置的日志记录器 

99 """ 

100 return _global_logger 

101 

102 

103def set_logger(logger: Logger) -> None: 

104 """设置全局日志实例. 

105 

106 Args: 

107 logger: 新的日志记录器实例 

108 """ 

109 global _global_logger 

110 _global_logger = logger