Coverage for src/qdrant_loader_mcp_server/utils/logging.py: 77%

1"""Centralized logging configuration for the MCP Server application.""" 

2 

3import logging 

4import os 

5import re 

6import sys 

7 

8import structlog 

9 

10 

11class QdrantVersionFilter(logging.Filter): 

12 """Filter to suppress Qdrant version check warnings.""" 

13 

14 def filter(self, record): 

15 return "Failed to obtain server version" not in str(record.msg) 

16 

17 

18class ApplicationFilter(logging.Filter): 

19 """Filter to only show logs from our application.""" 

20 

21 def filter(self, record): 

22 # Show logs from our application and related modules 

23 return ( 

24 record.name.startswith("mcp_server") 

25 or record.name.startswith("src.") 

26 or record.name == "uvicorn" 

27 or record.name == "fastapi" 

28 or record.name == "__main__" # Allow logs from main module 

29 or record.name == "asyncio" # Allow logs from asyncio 

30 or record.name == "main" # Allow logs when started as a script 

31 or record.name == "qdrant_loader_mcp_server" # Allow logs from the package 

32 ) 

33 

34 

35class CleanFormatter(logging.Formatter): 

36 """Formatter that removes ANSI color codes.""" 

37 

38 def format(self, record): 

39 # Get the formatted message 

40 message = super().format(record) 

41 # Remove ANSI color codes 

42 ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") 

43 return ansi_escape.sub("", message) 

44 

45 

46try: 

47 # Use core logging config if available 

48 from qdrant_loader_core.logging import ( 

49 LoggingConfig as CoreLoggingConfig, # type: ignore 

50 ) 

51except Exception: # pragma: no cover - core may not be available 

52 CoreLoggingConfig = None # type: ignore 

53 

54 

55class LoggingConfig: 

56 """Wrapper that standardizes env handling and tracks current config. 

57 

58 Delegates to core LoggingConfig when available, while maintaining 

59 _initialized and _current_config for MCP server tests and utilities. 

60 """ 

61 

62 _initialized = False 

63 _current_config: tuple[str, str, str | None, bool] | None = None 

64 

65 @classmethod 

66 def setup( 

67 cls, 

68 level: str = "INFO", 

69 format: str = "console", 

70 file: str | None = None, 

71 suppress_qdrant_warnings: bool = True, 

72 ) -> None: 

73 # Resolve from environment when present for level; for file only when using all defaults 

74 env_level = os.getenv("MCP_LOG_LEVEL") 

75 resolved_level = (env_level or level).upper() 

76 env_file = os.getenv("MCP_LOG_FILE") 

77 all_defaults = ( 

78 level == "INFO" 

79 and format == "console" 

80 and file is None 

81 and suppress_qdrant_warnings is True 

82 ) 

83 resolved_file = ( 

84 file if file is not None else (env_file if all_defaults else None) 

85 ) 

86 disable_console_logging = ( 

87 os.getenv("MCP_DISABLE_CONSOLE_LOGGING", "").lower() == "true" 

88 ) 

89 

90 # Validate level 

91 if not hasattr(logging, resolved_level): 

92 raise ValueError(f"Invalid log level: {resolved_level}") 

93 

94 numeric_level = getattr(logging, resolved_level) 

95 

96 if CoreLoggingConfig is not None: 

97 # Delegate to core implementation 

98 CoreLoggingConfig.setup( 

99 level=resolved_level, 

100 format=format, 

101 file=resolved_file, 

102 suppress_qdrant_warnings=suppress_qdrant_warnings, 

103 disable_console=disable_console_logging, 

104 ) 

105 else: 

106 # Minimal fallback behavior 

107 handlers: list[logging.Handler] = [] 

108 if not disable_console_logging: 

109 stderr_handler = logging.StreamHandler(sys.stderr) 

110 stderr_handler.setFormatter(logging.Formatter("%(message)s")) 

111 handlers.append(stderr_handler) 

112 if resolved_file: 

113 file_handler = logging.FileHandler(resolved_file) 

114 file_handler.setFormatter(CleanFormatter("%(message)s")) 

115 handlers.append(file_handler) 

116 logging.basicConfig(level=numeric_level, handlers=handlers, force=True) 

117 if suppress_qdrant_warnings: 

118 logging.getLogger("qdrant_client").addFilter(QdrantVersionFilter()) 

119 

120 cls._initialized = True 

121 cls._current_config = ( 

122 resolved_level, 

123 format, 

124 resolved_file, 

125 suppress_qdrant_warnings, 

126 ) 

127 

128 @classmethod 

129 def get_logger(cls, name: str | None = None): # type: ignore 

130 if not cls._initialized: 

131 cls.setup() 

132 return structlog.get_logger(name)