Coverage for faststream / _internal / logger / logging.py: 97%

1import logging (empty)

2import sys (empty)

3from collections.abc import Mapping (empty)

4from logging import LogRecord (empty)

5from typing import TYPE_CHECKING, TextIO (empty)

6 

7from .formatter import ColourizedFormatter (empty)

8 

9if TYPE_CHECKING: (empty)

10 from faststream._internal.context.repository import ContextRepo 

11 

12 

13class ExtendedFilter(logging.Filter): (empty)

14 def __init__( (empty)

15 self, 

16 default_context: Mapping[str, str], 

17 message_id_ln: int, 

18 context: "ContextRepo", 

19 name: str = "", 

20 ) -> None: 

21 self.default_context = default_context (empty)

22 self.message_id_ln = message_id_ln (empty)

23 self.context = context (empty)

24 super().__init__(name) (empty)

25 

26 def filter(self, record: LogRecord) -> bool: (empty)

27 if is_suitable := super().filter(record): 27 ↛ 39line 27 didn't jump to line 39 because the condition on line 27 was always true(empty)

28 log_context: Mapping[str, str] = self.context.get_local( (empty)

29 "log_context", 

30 self.default_context, 

31 ) 

32 

33 for k, v in log_context.items(): (empty)

34 value = getattr(record, k, v) (empty)

35 setattr(record, k, value) (empty)

36 

37 record.message_id = getattr(record, "message_id", "")[: self.message_id_ln] (empty)

38 

39 return is_suitable (empty)

40 

41 

42def get_broker_logger( (empty)

43 name: str, 

44 default_context: Mapping[str, str], 

45 message_id_ln: int, 

46 fmt: str, 

47 context: "ContextRepo", 

48 log_level: int, 

49) -> logging.Logger: 

50 logger = get_logger( (empty)

51 name=f"faststream.access.{name}", 

52 log_level=log_level, 

53 stream=sys.stdout, 

54 fmt=fmt, 

55 ) 

56 logger.addFilter(ExtendedFilter(default_context, message_id_ln, context=context)) (empty)

57 return logger (empty)

58 

59 

60def get_logger(name: str, log_level: int, stream: "TextIO", fmt: str) -> logging.Logger: (empty)

61 logger = logging.getLogger(name) (empty)

62 logger.setLevel(log_level) (empty)

63 logger.propagate = False (empty)

64 set_logger_fmt(logger, stream=stream, fmt=fmt) (empty)

65 return logger (empty)

66 

67 

68def set_logger_fmt( (empty)

69 logger: logging.Logger, 

70 stream: "TextIO", 

71 fmt: str, 

72) -> None: 

73 if _handler_exists(logger): (empty)

74 return (empty)

75 

76 handler = logging.StreamHandler(stream=stream) (empty)

77 handler.setFormatter( (empty)

78 ColourizedFormatter( 

79 fmt=fmt, 

80 use_colors=True, 

81 ), 

82 ) 

83 logger.addHandler(handler) (empty)

84 

85 

86def _handler_exists(logger: logging.Logger) -> bool: (empty)

87 # Check if a StreamHandler for sys.stdout already exists in the logger. 

88 for handler in logger.handlers: (empty)

89 if isinstance(handler, logging.StreamHandler) and handler.stream == sys.stdout: (empty)

90 return True (empty)

91 return False (empty)

92 

93 

94logger = get_logger( (empty)

95 name="faststream", 

96 log_level=logging.INFO, 

97 stream=sys.stderr, 

98 fmt="%(asctime)s %(levelname)8s - %(message)s", 

99)