Coverage for faststream / _internal / logger / params_storage.py: 88%

1import logging (empty)

2from abc import abstractmethod (empty)

3from typing import TYPE_CHECKING, Any, Optional, Protocol (empty)

4from weakref import WeakSet (empty)

5 

6from faststream._internal.constants import EMPTY (empty)

7 

8if TYPE_CHECKING: (empty)

9 from faststream._internal.basic_types import LoggerProto 

10 from faststream._internal.context import ContextRepo 

11 

12 

13def make_logger_storage( (empty)

14 logger: Optional["LoggerProto"], 

15 default_storage_cls: type["DefaultLoggerStorage"], 

16) -> "LoggerParamsStorage": 

17 if logger is EMPTY: (empty)

18 return default_storage_cls() (empty)

19 

20 return EmptyLoggerStorage() if logger is None else ManualLoggerStorage(logger) (empty)

21 

22 

23class LoggerParamsStorage(Protocol): (empty)

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

25 

26 def get_logger(self, *, context: "ContextRepo") -> Optional["LoggerProto"]: ... (empty)

27 

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

29 

30 

31class EmptyLoggerStorage(LoggerParamsStorage): (empty)

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

33 pass (empty)

34 

35 def get_logger(self, *, context: "ContextRepo") -> None: (empty)

36 return None (empty)

37 

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

39 pass (empty)

40 

41 

42class ManualLoggerStorage(LoggerParamsStorage): (empty)

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

44 self.__logger = logger (empty)

45 

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

47 pass (empty)

48 

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

50 return self.__logger (empty)

51 

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

53 if getattr(self.__logger, "setLevel", None): 

54 self.__logger.setLevel(level) # type: ignore[attr-defined] 

55 

56 

57class DefaultLoggerStorage(LoggerParamsStorage): (empty)

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

59 # will be used to build logger in `get_logger` method 

60 self.logger_log_level = logging.INFO (empty)

61 

62 self._logger_ref = WeakSet[logging.Logger]() (empty)

63 

64 @abstractmethod (empty)

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

66 raise NotImplementedError 

67 

68 def _get_logger_ref(self) -> logging.Logger | None: (empty)

69 if self._logger_ref: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true(empty)

70 return next(iter(self._logger_ref)) 

71 

72 return None (empty)

73 

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

75 if lg := self._get_logger_ref(): (empty)

76 lg.setLevel(level) 

77 

78 self.logger_log_level = level (empty)