Coverage for faststream / redis / subscriber / usecases / list_subscriber.py: 95%

1from collections.abc import AsyncIterator (empty)

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

3 

4import anyio (empty)

5from typing_extensions import override (empty)

6 

7from faststream._internal.endpoint.subscriber.mixins import ConcurrentMixin (empty)

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

9from faststream.redis.message import ( (empty)

10 BatchListMessage, 

11 DefaultListMessage, 

12 RedisListMessage, 

13) 

14from faststream.redis.parser import ( (empty)

15 RedisBatchListParser, 

16 RedisListParser, 

17) 

18 

19from .basic import LogicSubscriber (empty)

20 

21if TYPE_CHECKING: (empty)

22 from redis.asyncio.client import Redis 

23 

24 from faststream._internal.endpoint.subscriber import SubscriberSpecification 

25 from faststream._internal.endpoint.subscriber.call_item import ( 

26 CallsCollection, 

27 ) 

28 from faststream.message import StreamMessage as BrokerStreamMessage 

29 from faststream.redis.schemas import ListSub 

30 from faststream.redis.subscriber.config import RedisSubscriberConfig 

31 from faststream.redis.subscriber.specification import RedisSubscriberSpecification 

32 

33TopicName: TypeAlias = bytes (empty)

34Offset: TypeAlias = bytes (empty)

35 

36 

37class _ListHandlerMixin(LogicSubscriber): (empty)

38 def __init__( (empty)

39 self, 

40 config: "RedisSubscriberConfig", 

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

42 calls: "CallsCollection[Any]", 

43 ) -> None: 

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

45 self._read_lock = anyio.Lock() (empty)

46 assert config.list_sub (empty)

47 self._list_sub = config.list_sub (empty)

48 

49 @property (empty)

50 def list_sub(self) -> "ListSub": (empty)

51 return self._list_sub.add_prefix(self._outer_config.prefix) (empty)

52 

53 def get_log_context( (empty)

54 self, 

55 message: Optional["BrokerStreamMessage[Any]"], 

56 ) -> dict[str, str]: 

57 return self.build_log_context( (empty)

58 message=message, 

59 channel=self.list_sub.name, 

60 ) 

61 

62 @override (empty)

63 async def _consume( # type: ignore[override] (empty)

64 self, 

65 client: "Redis[bytes]", 

66 *, 

67 start_signal: "anyio.Event", 

68 ) -> None: 

69 if await client.ping(): 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was always true(empty)

70 start_signal.set() (empty)

71 await super()._consume(client, start_signal=start_signal) (empty)

72 

73 @override (empty)

74 async def start(self) -> None: (empty)

75 await super().start(self._client) (empty)

76 

77 @override (empty)

78 async def stop(self) -> None: (empty)

79 with anyio.move_on_after(self._outer_config.graceful_timeout): (empty)

80 async with self._read_lock: (empty)

81 await super().stop() (empty)

82 

83 @override (empty)

84 async def get_one( (empty)

85 self, 

86 *, 

87 timeout: float = 5.0, 

88 ) -> "RedisListMessage | None": 

89 assert not self.calls, ( (empty)

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

91 ) 

92 

93 sleep_interval = timeout / 10 (empty)

94 raw_message = None (empty)

95 

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

97 while ( # noqa: ASYNC110 (empty)

98 raw_message := await self._client.lpop(name=self.list_sub.name) 

99 ) is None: 

100 await anyio.sleep(sleep_interval) (empty)

101 

102 if not raw_message: (empty)

103 return None (empty)

104 

105 redis_incoming_msg = DefaultListMessage( (empty)

106 type="list", 

107 data=raw_message, 

108 channel=self.list_sub.name, 

109 ) 

110 

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

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

113 

114 msg: RedisListMessage = await process_msg( # type: ignore[assignment] (empty)

115 msg=redis_incoming_msg, 

116 middlewares=( 

117 m(redis_incoming_msg, context=context) for m in self._broker_middlewares 

118 ), 

119 parser=async_parser, 

120 decoder=async_decoder, 

121 ) 

122 return msg (empty)

123 

124 @override (empty)

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

126 assert not self.calls, ( (empty)

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

128 ) 

129 

130 timeout = 5 (empty)

131 sleep_interval = timeout / 10 (empty)

132 raw_message = None (empty)

133 

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

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

136 

137 while True: (empty)

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

139 while ( # noqa: ASYNC110 139 ↛ 142line 139 didn't jump to line 142 because the condition on line 139 was never true(empty)

140 raw_message := await self._client.lpop(name=self.list_sub.name) 

141 ) is None: 

142 await anyio.sleep(sleep_interval) 

143 

144 if not raw_message: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true(empty)

145 continue 

146 

147 redis_incoming_msg = DefaultListMessage( (empty)

148 type="list", 

149 data=raw_message, 

150 channel=self.list_sub.name, 

151 ) 

152 

153 msg: RedisListMessage = await process_msg( # type: ignore[assignment] (empty)

154 msg=redis_incoming_msg, 

155 middlewares=( 

156 m(redis_incoming_msg, context=context) 

157 for m in self._broker_middlewares 

158 ), 

159 parser=async_parser, 

160 decoder=async_decoder, 

161 ) 

162 yield msg (empty)

163 

164 

165class ListSubscriber(_ListHandlerMixin): (empty)

166 def __init__( (empty)

167 self, 

168 config: "RedisSubscriberConfig", 

169 specification: "RedisSubscriberSpecification", 

170 calls: "CallsCollection[Any]", 

171 ) -> None: 

172 parser = RedisListParser(config) (empty)

173 config.parser = parser.parse_message (empty)

174 config.decoder = parser.decode_message (empty)

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

176 

177 async def _get_msgs(self, client: "Redis[bytes]") -> None: (empty)

178 async with self._read_lock: (empty)

179 raw_msg = await client.blpop( (empty)

180 self.list_sub.name, 

181 timeout=self.list_sub.polling_interval, 

182 ) 

183 

184 if raw_msg: (empty)

185 _, msg_data = raw_msg (empty)

186 

187 msg = DefaultListMessage( (empty)

188 type="list", 

189 data=msg_data, 

190 channel=self.list_sub.name, 

191 ) 

192 

193 await self.consume_one(msg) (empty)

194 

195 

196class ListBatchSubscriber(_ListHandlerMixin): (empty)

197 def __init__( (empty)

198 self, 

199 config: "RedisSubscriberConfig", 

200 specification: "RedisSubscriberSpecification", 

201 calls: "CallsCollection[Any]", 

202 ) -> None: 

203 parser = RedisBatchListParser(config) (empty)

204 config.parser = parser.parse_message (empty)

205 config.decoder = parser.decode_message (empty)

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

207 

208 async def _get_msgs(self, client: "Redis[bytes]") -> None: (empty)

209 async with self._read_lock: (empty)

210 raw_msgs = await client.lpop( (empty)

211 name=self.list_sub.name, 

212 count=self.list_sub.max_records, 

213 ) 

214 

215 if raw_msgs: (empty)

216 msg = BatchListMessage( (empty)

217 type="blist", 

218 channel=self.list_sub.name, 

219 data=raw_msgs, 

220 ) 

221 

222 await self.consume_one(msg) (empty)

223 

224 if not raw_msgs: (empty)

225 await anyio.sleep(self.list_sub.polling_interval) (empty)

226 

227 

228class ListConcurrentSubscriber( (empty)

229 ConcurrentMixin["BrokerStreamMessage[Any]"], 

230 ListSubscriber, 

231): 

232 async def start(self) -> None: (empty)

233 await super().start() (empty)

234 self.start_consume_task() (empty)

235 

236 async def consume_one(self, msg: "BrokerStreamMessage[Any]") -> None: (empty)

237 await self._put_msg(msg) (empty)