Coverage for faststream / message / message.py: 96%

1from enum import Enum (empty)

2from typing import ( (empty)

3 TYPE_CHECKING, 

4 Any, 

5 Generic, 

6 Optional, 

7 TypeVar, 

8) 

9from uuid import uuid4 (empty)

10 

11from .source_type import SourceType (empty)

12 

13if TYPE_CHECKING: (empty)

14 from faststream._internal.types import AsyncCallable 

15 

16# prevent circular imports 

17MsgType = TypeVar("MsgType") (empty)

18 

19_NOT_CACHED = object() (empty)

20 

21 

22class AckStatus(str, Enum): (empty)

23 ACKED = "ACKED" (empty)

24 NACKED = "NACKED" (empty)

25 REJECTED = "REJECTED" (empty)

26 

27 

28class StreamMessage(Generic[MsgType]): (empty)

29 """Generic class to represent a stream message.""" 

30 

31 def __init__( (empty)

32 self, 

33 raw_message: "MsgType", 

34 body: bytes | Any, 

35 *, 

36 headers: dict[str, Any] | None = None, 

37 reply_to: str = "", 

38 batch_headers: list[dict[str, Any]] | None = None, 

39 path: dict[str, Any] | None = None, 

40 content_type: str | None = None, 

41 correlation_id: str | None = None, 

42 message_id: str | None = None, 

43 source_type: SourceType = SourceType.CONSUME, 

44 ) -> None: 

45 self.raw_message = raw_message (empty)

46 self.body = body (empty)

47 self.reply_to = reply_to (empty)

48 self.content_type = content_type (empty)

49 self.source_type = source_type (empty)

50 

51 self.headers = headers or {} (empty)

52 self.batch_headers = batch_headers or [] (empty)

53 self.path = path or {} (empty)

54 self.correlation_id = correlation_id or str(uuid4()) (empty)

55 self.message_id = message_id or self.correlation_id (empty)

56 

57 self.committed: AckStatus | None = None (empty)

58 self.processed = False (empty)

59 

60 # Setup later 

61 self.__decoder: AsyncCallable | None = None (empty)

62 self.__decoded_caches: dict[ (empty)

63 Any, 

64 Any, 

65 ] = {} # Cache values between filters and tests 

66 

67 def set_decoder(self, decoder: "AsyncCallable") -> None: (empty)

68 self.__decoder = decoder (empty)

69 

70 def clear_cache(self) -> None: (empty)

71 self.__decoded_caches.clear() 

72 

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

74 inner = ", ".join( (empty)

75 filter( (empty)

76 bool, (empty)

77 ( (empty)

78 f"body={self.body!r}", (empty)

79 f"content_type={self.content_type}", (empty)

80 f"message_id={self.message_id}", (empty)

81 f"correlation_id={self.correlation_id}", (empty)

82 f"reply_to={self.reply_to}" if self.reply_to else "", (empty)

83 f"headers={self.headers}", (empty)

84 f"path={self.path}", (empty)

85 f"committed={self.committed}", (empty)

86 f"raw_message={self.raw_message}", (empty)

87 ), 

88 ), 

89 ) 

90 

91 return f"{self.__class__.__name__}({inner})" (empty)

92 

93 async def decode(self) -> Optional["Any"]: (empty)

94 """Serialize the message by lazy decoder. 

95 

96 Returns a cache after first usage. To prevent such behavior, please call 

97 `message.clear_cache()` after `message.body` changes. 

98 """ 

99 assert self.__decoder, "You should call `set_decoder()` method first." (empty)

100 

101 if ( (empty)

102 result := self.__decoded_caches.get(self.__decoder, _NOT_CACHED) 

103 ) is _NOT_CACHED: 

104 result = self.__decoded_caches[self.__decoder] = await self.__decoder(self) (empty)

105 

106 return result (empty)

107 

108 async def ack(self) -> None: (empty)

109 if self.committed is None: (empty)

110 self.committed = AckStatus.ACKED (empty)

111 

112 async def nack(self) -> None: (empty)

113 if self.committed is None: 113 ↛ exitline 113 didn't return from function 'nack' because the condition on line 113 was always true(empty)

114 self.committed = AckStatus.NACKED (empty)

115 

116 async def reject(self) -> None: (empty)

117 if self.committed is None: (empty)

118 self.committed = AckStatus.REJECTED (empty)