Coverage for faststream / nats / subscriber / usecases / key_value_subscriber.py: 87%

1from collections.abc import AsyncIterator, Iterable (empty)

2from contextlib import suppress (empty)

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

4 

5import anyio (empty)

6from nats.errors import ConnectionClosedError, TimeoutError (empty)

7from typing_extensions import override (empty)

8 

9from faststream._internal.endpoint.subscriber.mixins import TasksMixin (empty)

10from faststream._internal.endpoint.utils import process_msg (empty)

11from faststream.nats.parser import KvParser (empty)

12from faststream.nats.subscriber.adapters import UnsubscribeAdapter (empty)

13 

14from .basic import LogicSubscriber (empty)

15 

16if TYPE_CHECKING: (empty)

17 from nats.js.kv import KeyValue 

18 

19 from faststream._internal.endpoint.publisher import PublisherProto 

20 from faststream._internal.endpoint.subscriber import SubscriberSpecification 

21 from faststream._internal.endpoint.subscriber.call_item import CallsCollection 

22 from faststream.message import StreamMessage 

23 from faststream.nats.message import NatsKvMessage 

24 from faststream.nats.schemas import KvWatch 

25 from faststream.nats.subscriber.config import NatsSubscriberConfig 

26 

27 

28class KeyValueWatchSubscriber( (empty)

29 TasksMixin, 

30 LogicSubscriber["KeyValue.Entry"], 

31): 

32 subscription: Optional["UnsubscribeAdapter[KeyValue.KeyWatcher]"] (empty)

33 _fetch_sub: UnsubscribeAdapter["KeyValue.KeyWatcher"] | None (empty)

34 

35 def __init__( (empty)

36 self, 

37 config: "NatsSubscriberConfig", 

38 specification: "SubscriberSpecification[Any, Any]", 

39 calls: "CallsCollection[KeyValue.Entry]", 

40 *, 

41 kv_watch: "KvWatch", 

42 ) -> None: 

43 parser = KvParser(pattern=config.subject) (empty)

44 config.decoder = parser.decode_message (empty)

45 config.parser = parser.parse_message (empty)

46 super().__init__(config, specification, calls) (empty)

47 

48 self.kv_watch = kv_watch (empty)

49 

50 @override (empty)

51 async def get_one( (empty)

52 self, 

53 *, 

54 timeout: float = 5, 

55 ) -> Optional["NatsKvMessage"]: 

56 assert not self.calls, ( (empty)

57 "You can't use `get_one` method if subscriber has registered handlers." 

58 ) 

59 

60 if not self._fetch_sub: 60 ↛ 76line 60 didn't jump to line 76 because the condition on line 60 was always true(empty)

61 bucket = await self._outer_config.kv_declarer.create_key_value( (empty)

62 bucket=self.kv_watch.name, 

63 declare=self.kv_watch.declare, 

64 ) 

65 

66 fetch_sub = self._fetch_sub = UnsubscribeAdapter["KeyValue.KeyWatcher"]( (empty)

67 await bucket.watch( 

68 keys=self.clear_subject, 

69 headers_only=self.kv_watch.headers_only, 

70 include_history=self.kv_watch.include_history, 

71 ignore_deletes=self.kv_watch.ignore_deletes, 

72 meta_only=self.kv_watch.meta_only, 

73 ), 

74 ) 

75 else: 

76 fetch_sub = self._fetch_sub 

77 

78 msg = None (empty)

79 sleep_interval = timeout / 10 (empty)

80 with anyio.move_on_after(timeout): (empty)

81 while ( # noqa: ASYNC110 (empty)

82 msg := await fetch_sub.obj.updates(timeout) # type: ignore[no-untyped-call] 

83 ) is None: 

84 await anyio.sleep(sleep_interval) (empty)

85 

86 context = self._outer_config.fd_config.context (empty)

87 async_parser, async_decoder = self._get_parser_and_decoder() (empty)

88 

89 return cast( (empty)

90 "NatsKvMessage", 

91 await process_msg( 

92 msg=msg, 

93 middlewares=(m(msg, context=context) for m in self._broker_middlewares), 

94 parser=async_parser, 

95 decoder=async_decoder, 

96 ), 

97 ) 

98 

99 @override (empty)

100 async def __aiter__(self) -> AsyncIterator["NatsKvMessage"]: # type: ignore[override] (empty)

101 assert not self.calls, ( (empty)

102 "You can't use iterator if subscriber has registered handlers." 

103 ) 

104 

105 if not self._fetch_sub: 105 ↛ 121line 105 didn't jump to line 121 because the condition on line 105 was always true(empty)

106 bucket = await self._outer_config.kv_declarer.create_key_value( (empty)

107 bucket=self.kv_watch.name, 

108 declare=self.kv_watch.declare, 

109 ) 

110 

111 fetch_sub = self._fetch_sub = UnsubscribeAdapter["KeyValue.KeyWatcher"]( (empty)

112 await bucket.watch( 

113 keys=self.clear_subject, 

114 headers_only=self.kv_watch.headers_only, 

115 include_history=self.kv_watch.include_history, 

116 ignore_deletes=self.kv_watch.ignore_deletes, 

117 meta_only=self.kv_watch.meta_only, 

118 ), 

119 ) 

120 else: 

121 fetch_sub = self._fetch_sub 

122 

123 timeout = 5 (empty)

124 sleep_interval = timeout / 10 (empty)

125 

126 context = self._outer_config.fd_config.context (empty)

127 async_parser, async_decoder = self._get_parser_and_decoder() (empty)

128 

129 while True: (empty)

130 msg = None (empty)

131 with anyio.move_on_after(timeout): (empty)

132 while ( # noqa: ASYNC110 (empty)

133 msg := await fetch_sub.obj.updates(timeout) # type: ignore[no-untyped-call] 

134 ) is None: 

135 await anyio.sleep(sleep_interval) (empty)

136 

137 if msg is None: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true(empty)

138 continue 

139 

140 yield cast( (empty)

141 "NatsKvMessage", 

142 await process_msg( 

143 msg=msg, 

144 middlewares=( 

145 m(msg, context=context) for m in self._broker_middlewares 

146 ), 

147 parser=async_parser, 

148 decoder=async_decoder, 

149 ), 

150 ) 

151 

152 @override (empty)

153 async def _create_subscription(self) -> None: (empty)

154 if self.subscription: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true(empty)

155 return 

156 

157 bucket = await self._outer_config.kv_declarer.create_key_value( (empty)

158 bucket=self.kv_watch.name, 

159 declare=self.kv_watch.declare, 

160 ) 

161 

162 self.subscription = UnsubscribeAdapter["KeyValue.KeyWatcher"]( (empty)

163 await bucket.watch( 

164 keys=self.clear_subject, 

165 headers_only=self.kv_watch.headers_only, 

166 include_history=self.kv_watch.include_history, 

167 ignore_deletes=self.kv_watch.ignore_deletes, 

168 meta_only=self.kv_watch.meta_only, 

169 ), 

170 ) 

171 

172 self.add_task(self.__consume_watch) (empty)

173 

174 async def __consume_watch(self) -> None: (empty)

175 assert self.subscription, "You should call `create_subscription` at first." (empty)

176 

177 key_watcher = self.subscription.obj (empty)

178 

179 while self.running: 179 ↛ exitline 179 didn't return from function '__consume_watch' because the condition on line 179 was always true(empty)

180 with suppress(ConnectionClosedError, TimeoutError): (empty)

181 message = cast( (empty)

182 "KeyValue.Entry | None", 

183 await key_watcher.updates(self.kv_watch.timeout), # type: ignore[no-untyped-call] 

184 ) 

185 

186 if message: (empty)

187 await self.consume(message) (empty)

188 

189 def _make_response_publisher( (empty)

190 self, 

191 message: "StreamMessage[KeyValue.Entry]", 

192 ) -> Iterable["PublisherProto"]: 

193 """Create Publisher objects to use it as one of `publishers` in `self.consume` scope. 

194 

195 Args: 

196 message: Message requiring reply 

197 """ 

198 return () 

199 

200 def get_log_context( (empty)

201 self, 

202 message: Optional["StreamMessage[KeyValue.Entry]"], 

203 ) -> dict[str, str]: 

204 """Log context factory using in `self.consume` scope. 

205 

206 Args: 

207 message: Message which we are building context for 

208 """ 

209 return self.build_log_context( (empty)

210 message=message, 

211 subject=self.subject, 

212 stream=self.kv_watch.name, 

213 )