Coverage for faststream / nats / broker / logging.py: 95%

1import logging (empty)

2from functools import partial (empty)

3from typing import TYPE_CHECKING, Any (empty)

4 

5from faststream._internal.logger import DefaultLoggerStorage, make_logger_state (empty)

6from faststream._internal.logger.logging import get_broker_logger (empty)

7 

8if TYPE_CHECKING: (empty)

9 from faststream._internal.basic_types import LoggerProto 

10 from faststream._internal.context import ContextRepo 

11 

12 

13class NatsParamsStorage(DefaultLoggerStorage): (empty)

14 def __init__(self) -> None: (empty)

15 super().__init__() (empty)

16 

17 self._max_queue_len = 0 (empty)

18 self._max_stream_len = 0 (empty)

19 self._max_subject_len = 4 (empty)

20 

21 self.logger_log_level = logging.INFO (empty)

22 

23 def set_level(self, level: int) -> None: (empty)

24 self.logger_log_level = level 

25 

26 def register_subscriber(self, params: dict[str, Any]) -> None: (empty)

27 self._max_subject_len = max( (empty)

28 ( 

29 self._max_subject_len, 

30 len(params.get("subject", "")), 

31 ), 

32 ) 

33 self._max_queue_len = max( (empty)

34 ( 

35 self._max_queue_len, 

36 len(params.get("queue", "")), 

37 ), 

38 ) 

39 self._max_stream_len = max( (empty)

40 ( 

41 self._max_stream_len, 

42 len(params.get("stream", "")), 

43 ), 

44 ) 

45 

46 def get_logger(self, *, context: "ContextRepo") -> "LoggerProto": (empty)

47 # TODO: generate unique logger names to not share between brokers 

48 if not (lg := self._get_logger_ref()): 48 ↛ 80line 48 didn't jump to line 80 because the condition on line 48 was always true(empty)

49 message_id_ln = 10 (empty)

50 

51 lg = get_broker_logger( (empty)

52 name="nats", 

53 default_context={ 

54 "subject": "", 

55 "stream": "", 

56 "queue": "", 

57 }, 

58 message_id_ln=message_id_ln, 

59 fmt="".join(( 

60 "%(asctime)s %(levelname)-8s - ", 

61 ( 

62 f"%(stream)-{self._max_stream_len}s | " 

63 if self._max_stream_len 

64 else "" 

65 ), 

66 ( 

67 f"%(queue)-{self._max_queue_len}s | " 

68 if self._max_queue_len 

69 else "" 

70 ), 

71 f"%(subject)-{self._max_subject_len}s | ", 

72 f"%(message_id)-{message_id_ln}s - ", 

73 "%(message)s", 

74 )), 

75 context=context, 

76 log_level=self.logger_log_level, 

77 ) 

78 self._logger_ref.add(lg) (empty)

79 

80 return lg (empty)

81 

82 

83make_nats_logger_state = partial( (empty)

84 make_logger_state, 

85 default_storage_cls=NatsParamsStorage, 

86)