Coverage for faststream / middlewares / logging.py: 95%

1import logging (empty)

2from typing import TYPE_CHECKING, Any, Optional (empty)

3 

4from faststream._internal.middlewares import BaseMiddleware (empty)

5from faststream.exceptions import IgnoredException (empty)

6from faststream.message.source_type import SourceType (empty)

7 

8if TYPE_CHECKING: (empty)

9 from types import TracebackType 

10 

11 from faststream._internal.basic_types import AsyncFuncAny 

12 from faststream._internal.context.repository import ContextRepo 

13 from faststream._internal.logger import LoggerState 

14 from faststream.message import StreamMessage 

15 

16 

17class CriticalLogMiddleware: (empty)

18 def __init__(self, logger: "LoggerState") -> None: (empty)

19 """Initialize the class.""" 

20 self.logger = logger (empty)

21 

22 def __call__( (empty)

23 self, 

24 msg: Any | None, 

25 /, 

26 *, 

27 context: "ContextRepo", 

28 ) -> "_LoggingMiddleware": 

29 return _LoggingMiddleware( (empty)

30 logger=self.logger, 

31 msg=msg, 

32 context=context, 

33 ) 

34 

35 

36class _LoggingMiddleware(BaseMiddleware): (empty)

37 """A middleware class for logging critical errors.""" 

38 

39 def __init__( (empty)

40 self, 

41 *, 

42 logger: "LoggerState", 

43 context: "ContextRepo", 

44 msg: Any | None, 

45 ) -> None: 

46 super().__init__(msg, context=context) (empty)

47 self.logger = logger (empty)

48 self.source_type = SourceType.CONSUME (empty)

49 

50 async def consume_scope( (empty)

51 self, 

52 call_next: "AsyncFuncAny", 

53 msg: "StreamMessage[Any]", 

54 ) -> Any: 

55 source_type = self.source_type = msg.source_type (empty)

56 

57 if source_type is not SourceType.RESPONSE: (empty)

58 self.logger.log( (empty)

59 "Received", 

60 extra=self.context.get_local("log_context", {}), 

61 ) 

62 

63 return await call_next(msg) (empty)

64 

65 async def __aexit__( (empty)

66 self, 

67 exc_type: type[BaseException] | None = None, 

68 exc_val: BaseException | None = None, 

69 exc_tb: Optional["TracebackType"] = None, 

70 ) -> bool: 

71 """Asynchronously called after processing.""" 

72 if self.source_type is not SourceType.RESPONSE: 72 ↛ 93line 72 didn't jump to line 93 because the condition on line 72 was always true(empty)

73 c = self.context.get_local("log_context", {}) (empty)

74 

75 if exc_type: (empty)

76 # TODO: move critical logging to `subscriber.consume()` method 

77 if issubclass(exc_type, IgnoredException): (empty)

78 self.logger.log( (empty)

79 message=str(exc_val), 

80 extra=c, 

81 ) 

82 

83 else: 

84 self.logger.log( (empty)

85 message=f"{exc_type.__name__}: {exc_val}", 

86 log_level=logging.ERROR, 

87 exc_info=exc_val, 

88 extra=c, 

89 ) 

90 

91 self.logger.log(message="Processed", extra=c) (empty)

92 

93 await super().__aexit__(exc_type, exc_val, exc_tb) (empty)

94 

95 # Exception was not processed 

96 return False (empty)