Coverage for snekql/structured_logging.py: 100%

1"""Structured logging contracts for snekql runtime observability.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6from typing import Protocol 

7 

8 

9class StructuredLogger(Protocol): 

10 """Structured logger accepted by the Query Runtime. 

11 

12 Any logger with structlog-style methods satisfies this protocol: 

13 

14 >>> class AppLogger: 

15 ... def debug(self, event: str, **fields: object) -> object: 

16 ... return None 

17 ... def info(self, event: str, **fields: object) -> object: 

18 ... return None 

19 ... def warning(self, event: str, **fields: object) -> object: 

20 ... return None 

21 ... def error(self, event: str, **fields: object) -> object: 

22 ... return None 

23 >>> logger: StructuredLogger = AppLogger() 

24 >>> logger.debug("query executed", sql="SELECT 1", params=()) 

25 """ 

26 

27 def debug(self, event: str, **fields: object) -> object: ... 

28 

29 def info(self, event: str, **fields: object) -> object: ... 

30 

31 def warning(self, event: str, **fields: object) -> object: ... 

32 

33 def error(self, event: str, **fields: object) -> object: ... 

34 

35 

36class ResolvedStructuredLogger(Protocol): 

37 """Internal logger shape after snekql wraps application loggers.""" 

38 

39 def debug(self, event: str, **fields: object) -> None: ... 

40 

41 def info(self, event: str, **fields: object) -> None: ... 

42 

43 def warning(self, event: str, **fields: object) -> None: ... 

44 

45 def error(self, event: str, **fields: object) -> None: ... 

46 

47 

48class _SafeStructuredLogger: 

49 """Best-effort logger wrapper that keeps observability out of control flow.""" 

50 

51 def __init__(self, *, logger: StructuredLogger) -> None: 

52 self.logger: StructuredLogger = logger 

53 

54 def debug(self, event: str, **fields: object) -> None: 

55 with contextlib.suppress(Exception): 

56 _ = self.logger.debug(event, **fields) 

57 

58 def info(self, event: str, **fields: object) -> None: 

59 with contextlib.suppress(Exception): 

60 _ = self.logger.info(event, **fields) 

61 

62 def warning(self, event: str, **fields: object) -> None: 

63 with contextlib.suppress(Exception): 

64 _ = self.logger.warning(event, **fields) 

65 

66 def error(self, event: str, **fields: object) -> None: 

67 with contextlib.suppress(Exception): 

68 _ = self.logger.error(event, **fields) 

69 

70 

71def resolve_structured_logger( 

72 *, 

73 logger: StructuredLogger, 

74) -> ResolvedStructuredLogger: 

75 """Return a safe structured logger for runtime internals. 

76 

77 snekql treats logging as best-effort observability. A user logger failure 

78 should not change Database, Transaction, or query execution behavior. 

79 """ 

80 

81 return _SafeStructuredLogger(logger=logger)