Coverage for faststream / _internal / logger / logger_proxy.py: 85%

1from abc import abstractmethod (empty)

2from collections.abc import Mapping (empty)

3from typing import Any, Optional (empty)

4 

5from faststream._internal.basic_types import LoggerProto (empty)

6from faststream.exceptions import IncorrectState (empty)

7 

8 

9class LoggerObject(LoggerProto): (empty)

10 logger: Optional["LoggerProto"] (empty)

11 

12 @abstractmethod (empty)

13 def __bool__(self) -> bool: ... (empty)

14 

15 

16class NotSetLoggerObject(LoggerObject): (empty)

17 """Default logger proxy for state. 

18 

19 Raises an error if user tries to log smth before state setup. 

20 """ 

21 

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

23 self.logger = None (empty)

24 

25 def __bool__(self) -> bool: (empty)

26 return False (empty)

27 

28 def __repr__(self) -> str: (empty)

29 return f"{self.__class__.__name__}()" 

30 

31 def log( (empty)

32 self, 

33 level: int, 

34 msg: Any, 

35 /, 

36 *, 

37 exc_info: Any = None, 

38 extra: Mapping[str, Any] | None = None, 

39 ) -> None: 

40 err_msg = ( 

41 "Logger object not set. Please, call `broker.connect()` or " 

42 "`_setup_logger_state` of parent broker state." 

43 ) 

44 raise IncorrectState(err_msg) 

45 

46 

47class EmptyLoggerObject(LoggerObject): (empty)

48 """Empty logger proxy for state. 

49 

50 Will be used if user setup `logger=None`. 

51 """ 

52 

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

54 self.logger = None (empty)

55 

56 def __bool__(self) -> bool: (empty)

57 return True 

58 

59 def __repr__(self) -> str: (empty)

60 return f"{self.__class__.__name__}()" 

61 

62 def log( (empty)

63 self, 

64 level: int, 

65 msg: Any, 

66 /, 

67 *, 

68 exc_info: Any = None, 

69 extra: Mapping[str, Any] | None = None, 

70 ) -> None: 

71 pass (empty)

72 

73 

74class RealLoggerObject(LoggerObject): (empty)

75 """Empty logger proxy for state. 

76 

77 Will be used if user setup custom `logger` (.params_storage.ManualLoggerStorage) 

78 or in default logger case (.params_storage.DefaultLoggerStorage). 

79 """ 

80 

81 logger: "LoggerProto" (empty)

82 

83 def __init__(self, logger: "LoggerProto") -> None: (empty)

84 self.logger = logger (empty)

85 

86 def __bool__(self) -> bool: (empty)

87 return True (empty)

88 

89 def __repr__(self) -> str: (empty)

90 return f"{self.__class__.__name__}(logger={self.logger})" 

91 

92 def log( (empty)

93 self, 

94 level: int, 

95 msg: Any, 

96 /, 

97 *, 

98 exc_info: Any = None, 

99 extra: Mapping[str, Any] | None = None, 

100 ) -> None: 

101 self.logger.log( (empty)

102 level, 

103 msg, 

104 extra=extra, 

105 exc_info=exc_info, 

106 )