Coverage for faststream / nats / parser.py: 94%

1from typing import TYPE_CHECKING, Any (empty)

2 

3from faststream._internal.utils.path import match_path (empty)

4from faststream.message import ( (empty)

5 StreamMessage, 

6 decode_message, 

7) 

8from faststream.nats.message import ( (empty)

9 NatsBatchMessage, 

10 NatsKvMessage, 

11 NatsMessage, 

12 NatsObjMessage, 

13) 

14from faststream.nats.schemas.js_stream import compile_nats_wildcard (empty)

15 

16if TYPE_CHECKING: (empty)

17 from nats.aio.msg import Msg 

18 from nats.js.api import ObjectInfo 

19 from nats.js.kv import KeyValue 

20 

21 from faststream._internal.basic_types import DecodedMessage 

22 

23 

24class NatsBaseParser: (empty)

25 """A class to parse NATS messages.""" 

26 

27 def __init__( (empty)

28 self, 

29 *, 

30 pattern: str, 

31 ) -> None: 

32 path_re, _ = compile_nats_wildcard(pattern) (empty)

33 self._path_re = path_re (empty)

34 

35 async def decode_message( (empty)

36 self, 

37 msg: "StreamMessage[Any]", 

38 ) -> "DecodedMessage": 

39 return decode_message(msg) (empty)

40 

41 

42class NatsParser(NatsBaseParser): (empty)

43 """A class to parse NATS core messages.""" 

44 

45 def __init__(self, *, pattern: str, is_ack_disabled: bool) -> None: (empty)

46 super().__init__(pattern=pattern) (empty)

47 

48 self.is_ack_disabled = is_ack_disabled (empty)

49 

50 async def parse_message( (empty)

51 self, 

52 message: "Msg", 

53 ) -> "StreamMessage[Msg]": 

54 path = match_path(self._path_re, message.subject) (empty)

55 

56 headers = message.header or {} (empty)

57 

58 if self.is_ack_disabled: 58 ↛ 61line 58 didn't jump to line 61 because the condition on line 58 was always true(empty)

59 message._ackd = True (empty)

60 

61 return NatsMessage( (empty)

62 raw_message=message, 

63 body=message.data, 

64 path=path, 

65 reply_to=message.reply, 

66 headers=headers, 

67 content_type=headers.get("content-type", ""), 

68 message_id=headers.get("message_id"), 

69 correlation_id=headers.get("correlation_id"), 

70 ) 

71 

72 

73class JsParser(NatsBaseParser): (empty)

74 """A class to parse NATS JS messages.""" 

75 

76 async def parse_message( (empty)

77 self, 

78 message: "Msg", 

79 ) -> "StreamMessage[Msg]": 

80 path = match_path(self._path_re, message.subject) (empty)

81 

82 headers = message.header or {} (empty)

83 

84 return NatsMessage( (empty)

85 raw_message=message, 

86 body=message.data, 

87 path=path, 

88 reply_to=headers.get("reply_to", ""), # differ from core 

89 headers=headers, 

90 content_type=headers.get("content-type"), 

91 message_id=headers.get("message_id"), 

92 correlation_id=headers.get("correlation_id"), 

93 ) 

94 

95 

96class BatchParser(JsParser): (empty)

97 """A class to parse NATS batch messages.""" 

98 

99 async def parse_batch( (empty)

100 self, 

101 message: list["Msg"], 

102 ) -> "StreamMessage[list[Msg]]": 

103 body: list[bytes] = [] (empty)

104 batch_headers: list[dict[str, str]] = [] (empty)

105 

106 if message: 106 ↛ 114line 106 didn't jump to line 114 because the condition on line 106 was always true(empty)

107 path = match_path(self._path_re, message[0].subject) (empty)

108 

109 for m in message: (empty)

110 batch_headers.append(m.headers or {}) (empty)

111 body.append(m.data) (empty)

112 

113 else: 

114 path = {} 

115 

116 headers = next(iter(batch_headers), {}) (empty)

117 

118 return NatsBatchMessage( (empty)

119 raw_message=message, 

120 body=body, 

121 path=path, 

122 headers=headers, 

123 batch_headers=batch_headers, 

124 ) 

125 

126 async def decode_batch( (empty)

127 self, 

128 msg: "StreamMessage[list[Msg]]", 

129 ) -> list["DecodedMessage"]: 

130 data: list[DecodedMessage] = [] (empty)

131 

132 for m in msg.raw_message: (empty)

133 one_msg = await self.parse_message(m) (empty)

134 data.append(decode_message(one_msg)) (empty)

135 

136 return data (empty)

137 

138 

139class KvParser(NatsBaseParser): (empty)

140 async def parse_message( (empty)

141 self, 

142 msg: "KeyValue.Entry", 

143 ) -> StreamMessage["KeyValue.Entry"]: 

144 return NatsKvMessage( (empty)

145 raw_message=msg, 

146 body=msg.value, 

147 path=match_path(self._path_re, msg.key), 

148 ) 

149 

150 

151class ObjParser(NatsBaseParser): (empty)

152 async def parse_message(self, msg: "ObjectInfo") -> StreamMessage["ObjectInfo"]: (empty)

153 return NatsObjMessage( (empty)

154 raw_message=msg, 

155 body=msg.name, 

156 )