Coverage for src / quber / utils / logging.py: 100%

1""" 

2Logging configuration with Logfire integration. 

3 

4LangSmith tracing lives in `quber.agents.langsmith_tracer` and posts runs 

5to LangSmith's native `/runs/multipart` endpoint with explicit 

6`inputs`/`outputs` fields — the same pattern as `~/.claude/hooks/stop_hook.sh`. 

7That keeps the LangSmith UI showing actual prompts and responses rather 

8than empty input/output panels (the OTLP path's schema-translation gap). 

9""" 

10 

11import logging 

12import sys 

13from typing import Any, Optional 

14 

15import logfire 

16from loguru import logger 

17 

18from quber.settings import get_settings 

19 

20 

21def setup_logging( 

22 service_name: str = "quber", 

23 enable_logfire: bool = True, 

24 log_level: str = "INFO", 

25 suppress_external: bool = True, 

26) -> None: 

27 """ 

28 Set up logging with loguru and optional logfire integration. 

29 

30 Args: 

31 service_name: Name for the service in logfire 

32 enable_logfire: Whether to enable logfire telemetry 

33 log_level: Logging level (DEBUG, INFO, WARNING, ERROR) 

34 suppress_external: Whether to suppress verbose external library logs 

35 """ 

36 if suppress_external: 

37 logging.getLogger("docling").setLevel(logging.WARNING) 

38 logging.getLogger("docling.document_converter").setLevel(logging.WARNING) 

39 logging.getLogger("docling_core").setLevel(logging.WARNING) 

40 logging.getLogger("httpx").setLevel(logging.WARNING) 

41 logging.getLogger("httpcore").setLevel(logging.WARNING) 

42 logging.getLogger("anthropic").setLevel(logging.WARNING) 

43 logging.getLogger("openai").setLevel(logging.WARNING) 

44 logging.getLogger("langsmith").setLevel(logging.WARNING) 

45 

46 import warnings 

47 

48 warnings.filterwarnings("ignore", category=DeprecationWarning, module="docling") 

49 

50 logger.remove() 

51 logger.add( 

52 sys.stderr, 

53 format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>", 

54 level=log_level, 

55 ) 

56 

57 logfire_token = get_settings().obs.logfire_token 

58 if enable_logfire and logfire_token: 

59 try: 

60 logfire.configure( 

61 service_name=service_name, 

62 token=logfire_token, 

63 console=False, 

64 ) 

65 logfire.instrument_anthropic() 

66 logfire.instrument_openai() 

67 logfire.instrument_pydantic_ai() 

68 

69 if log_level == "DEBUG": 

70 logger.info(f"Logfire telemetry enabled for service: {service_name}") 

71 except Exception as e: 

72 logger.warning(f"Failed to configure logfire: {e}") 

73 else: 

74 if not logfire_token: 

75 logger.debug("Logfire token not found, telemetry disabled") 

76 

77 

78def get_logger(name: Optional[str] = None) -> Any: 

79 """ 

80 Get a logger instance. 

81 

82 Args: 

83 name: Optional name for the logger context 

84 

85 Returns: 

86 Logger instance 

87 """ 

88 if name: 

89 return logger.bind(context=name) 

90 return logger