Coverage for faststream / redis / publisher / usecase.py: 92%

1from abc import abstractmethod (empty)

2from collections.abc import Iterable (empty)

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

4 

5from typing_extensions import override (empty)

6 

7from faststream._internal.endpoint.publisher import ( (empty)

8 PublisherSpecification, 

9 PublisherUsecase, 

10) 

11from faststream.message import gen_cor_id (empty)

12from faststream.redis.response import RedisPublishCommand (empty)

13from faststream.response.publish_type import PublishType (empty)

14 

15from .producer import RedisFastProducer (empty)

16 

17if TYPE_CHECKING: (empty)

18 from redis.asyncio.client import Pipeline 

19 

20 from faststream._internal.basic_types import SendableMessage 

21 from faststream._internal.types import PublisherMiddleware 

22 from faststream.redis.message import RedisChannelMessage 

23 from faststream.redis.schemas import ListSub, PubSub, StreamSub 

24 from faststream.response import PublishCommand 

25 

26 from .config import RedisPublisherConfig 

27 

28 

29class LogicPublisher(PublisherUsecase): (empty)

30 """A class to represent a Redis publisher.""" 

31 

32 def __init__( (empty)

33 self, 

34 config: "RedisPublisherConfig", 

35 specification: "PublisherSpecification[Any, Any]", 

36 ) -> None: 

37 super().__init__(config, specification) (empty)

38 

39 self.config = config (empty)

40 

41 self.reply_to = config.reply_to (empty)

42 self.headers = config.headers or {} (empty)

43 

44 self.producer = self.config._outer_config.producer (empty)

45 

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

47 await super().start() (empty)

48 

49 broker_producer = self.config._outer_config.producer (empty)

50 

51 self.producer = RedisFastProducer( (empty)

52 connection=self.config._outer_config.connection, 

53 parser=broker_producer._parser.custom_func, 

54 decoder=broker_producer._decoder.custom_func, 

55 message_format=self.config.message_format, 

56 serializer=self.config._outer_config.fd_config._serializer, 

57 ) 

58 

59 @abstractmethod (empty)

60 def subscriber_property(self, *, name_only: bool) -> dict[str, Any]: (empty)

61 raise NotImplementedError 

62 

63 

64class ChannelPublisher(LogicPublisher): (empty)

65 def __init__( (empty)

66 self, 

67 config: "RedisPublisherConfig", 

68 specification: "PublisherSpecification[Any, Any]", 

69 *, 

70 channel: "PubSub", 

71 ) -> None: 

72 super().__init__(config, specification) (empty)

73 

74 self._channel = channel (empty)

75 

76 @property (empty)

77 def channel(self) -> "PubSub": (empty)

78 return self._channel.add_prefix(self._outer_config.prefix) (empty)

79 

80 @override (empty)

81 def subscriber_property(self, *, name_only: bool) -> dict[str, Any]: (empty)

82 return { (empty)

83 "channel": self.channel.name if name_only else self.channel, 

84 "list": None, 

85 "stream": None, 

86 } 

87 

88 @override (empty)

89 async def publish( (empty)

90 self, 

91 message: "SendableMessage" = None, 

92 channel: str | None = None, 

93 reply_to: str = "", 

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

95 correlation_id: str | None = None, 

96 *, 

97 pipeline: Optional["Pipeline[bytes]"] = None, 

98 ) -> int: 

99 cmd = RedisPublishCommand( (empty)

100 message, 

101 channel=channel or self.channel.name, 

102 reply_to=reply_to or self.reply_to, 

103 headers=self.headers | (headers or {}), 

104 correlation_id=correlation_id or gen_cor_id(), 

105 pipeline=pipeline, 

106 _publish_type=PublishType.PUBLISH, 

107 message_format=self.config.message_format, 

108 ) 

109 result: int = await self._basic_publish( (empty)

110 cmd, 

111 producer=self.producer, 

112 _extra_middlewares=(), 

113 ) 

114 return result (empty)

115 

116 @override (empty)

117 async def _publish( (empty)

118 self, 

119 cmd: Union["PublishCommand", "RedisPublishCommand"], 

120 *, 

121 _extra_middlewares: Iterable["PublisherMiddleware"], 

122 ) -> None: 

123 """This method should be called in subscriber flow only.""" 

124 cmd = RedisPublishCommand.from_cmd(cmd, message_format=self.config.message_format) (empty)

125 

126 cmd.set_destination(channel=self.channel.name) (empty)

127 

128 cmd.add_headers(self.headers, override=False) (empty)

129 cmd.reply_to = cmd.reply_to or self.reply_to (empty)

130 

131 await self._basic_publish( (empty)

132 cmd, 

133 producer=self.producer, 

134 _extra_middlewares=_extra_middlewares, 

135 ) 

136 

137 @override (empty)

138 async def request( (empty)

139 self, 

140 message: "SendableMessage" = None, 

141 channel: str | None = None, 

142 *, 

143 correlation_id: str | None = None, 

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

145 timeout: float | None = 30.0, 

146 ) -> "RedisChannelMessage": 

147 cmd = RedisPublishCommand( (empty)

148 message, 

149 channel=channel or self.channel.name, 

150 headers=self.headers | (headers or {}), 

151 correlation_id=correlation_id or gen_cor_id(), 

152 timeout=timeout, 

153 _publish_type=PublishType.REQUEST, 

154 message_format=self.config.message_format, 

155 ) 

156 

157 msg: RedisChannelMessage = await self._basic_request( (empty)

158 cmd, 

159 producer=self.producer, 

160 ) 

161 return msg (empty)

162 

163 

164class ListPublisher(LogicPublisher): (empty)

165 def __init__( (empty)

166 self, 

167 config: "RedisPublisherConfig", 

168 specification: "PublisherSpecification[Any, Any]", 

169 *, 

170 list: "ListSub", 

171 ) -> None: 

172 super().__init__(config, specification) (empty)

173 

174 self._list = list (empty)

175 

176 @property (empty)

177 def list(self) -> "ListSub": (empty)

178 return self._list.add_prefix(self._outer_config.prefix) (empty)

179 

180 @override (empty)

181 def subscriber_property(self, *, name_only: bool) -> dict[str, Any]: (empty)

182 return { (empty)

183 "channel": None, 

184 "list": self.list.name if name_only else self.list, 

185 "stream": None, 

186 } 

187 

188 @override (empty)

189 async def publish( (empty)

190 self, 

191 message: "SendableMessage" = None, 

192 list: str | None = None, 

193 reply_to: str = "", 

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

195 correlation_id: str | None = None, 

196 *, 

197 pipeline: Optional["Pipeline[bytes]"] = None, 

198 ) -> int: 

199 cmd = RedisPublishCommand( (empty)

200 message, 

201 list=list or self.list.name, 

202 reply_to=reply_to or self.reply_to, 

203 headers=self.headers | (headers or {}), 

204 correlation_id=correlation_id or gen_cor_id(), 

205 pipeline=pipeline, 

206 _publish_type=PublishType.PUBLISH, 

207 message_format=self.config.message_format, 

208 ) 

209 

210 result: int = await self._basic_publish( (empty)

211 cmd, 

212 producer=self.producer, 

213 _extra_middlewares=(), 

214 ) 

215 return result (empty)

216 

217 @override (empty)

218 async def _publish( (empty)

219 self, 

220 cmd: Union["PublishCommand", "RedisPublishCommand"], 

221 *, 

222 _extra_middlewares: Iterable["PublisherMiddleware"], 

223 ) -> None: 

224 """This method should be called in subscriber flow only.""" 

225 cmd = RedisPublishCommand.from_cmd(cmd, message_format=self.config.message_format) (empty)

226 

227 cmd.set_destination(list=self.list.name) (empty)

228 

229 cmd.add_headers(self.headers, override=False) (empty)

230 cmd.reply_to = cmd.reply_to or self.reply_to (empty)

231 

232 await self._basic_publish( (empty)

233 cmd, 

234 producer=self.producer, 

235 _extra_middlewares=_extra_middlewares, 

236 ) 

237 

238 @override (empty)

239 async def request( (empty)

240 self, 

241 message: "SendableMessage" = None, 

242 list: str | None = None, 

243 *, 

244 correlation_id: str | None = None, 

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

246 timeout: float | None = 30.0, 

247 ) -> "RedisChannelMessage": 

248 cmd = RedisPublishCommand( 

249 message, 

250 list=list or self.list.name, 

251 headers=self.headers | (headers or {}), 

252 correlation_id=correlation_id or gen_cor_id(), 

253 timeout=timeout, 

254 _publish_type=PublishType.REQUEST, 

255 message_format=self.config.message_format, 

256 ) 

257 

258 msg: RedisChannelMessage = await self._basic_request( 

259 cmd, 

260 producer=self.producer, 

261 ) 

262 return msg 

263 

264 

265class ListBatchPublisher(ListPublisher): (empty)

266 @override (empty)

267 async def publish( # type: ignore[override] (empty)

268 self, 

269 *messages: "SendableMessage", 

270 list: str, 

271 correlation_id: str | None = None, 

272 reply_to: str = "", 

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

274 pipeline: Optional["Pipeline[bytes]"] = None, 

275 ) -> int: 

276 cmd = RedisPublishCommand( 

277 *messages, 

278 list=list or self.list.name, 

279 reply_to=reply_to or self.reply_to, 

280 headers=self.headers | (headers or {}), 

281 correlation_id=correlation_id or gen_cor_id(), 

282 pipeline=pipeline, 

283 _publish_type=PublishType.PUBLISH, 

284 message_format=self.config.message_format, 

285 ) 

286 

287 result: int = await self._basic_publish_batch( 

288 cmd, 

289 producer=self.producer, 

290 _extra_middlewares=(), 

291 ) 

292 return result 

293 

294 @override (empty)

295 async def _publish( (empty)

296 self, 

297 cmd: Union["PublishCommand", "RedisPublishCommand"], 

298 *, 

299 _extra_middlewares: Iterable["PublisherMiddleware"], 

300 ) -> None: 

301 """This method should be called in subscriber flow only.""" 

302 cmd = RedisPublishCommand.from_cmd( (empty)

303 cmd, batch=True, message_format=self.config.message_format 

304 ) 

305 

306 cmd.set_destination(list=self.list.name) (empty)

307 

308 cmd.add_headers(self.headers, override=False) (empty)

309 cmd.reply_to = cmd.reply_to or self.reply_to (empty)

310 

311 await self._basic_publish_batch( (empty)

312 cmd, 

313 producer=self.producer, 

314 _extra_middlewares=_extra_middlewares, 

315 ) 

316 

317 

318class StreamPublisher(LogicPublisher): (empty)

319 def __init__( (empty)

320 self, 

321 config: "RedisPublisherConfig", 

322 specification: "PublisherSpecification[Any, Any]", 

323 *, 

324 stream: "StreamSub", 

325 ) -> None: 

326 super().__init__(config, specification) (empty)

327 self._stream = stream (empty)

328 

329 @property (empty)

330 def stream(self) -> "StreamSub": (empty)

331 return self._stream.add_prefix(self._outer_config.prefix) (empty)

332 

333 @override (empty)

334 def subscriber_property(self, *, name_only: bool) -> dict[str, Any]: (empty)

335 return { (empty)

336 "channel": None, 

337 "list": None, 

338 "stream": self.stream.name if name_only else self.stream, 

339 } 

340 

341 @override (empty)

342 async def publish( (empty)

343 self, 

344 message: "SendableMessage" = None, 

345 stream: str | None = None, 

346 reply_to: str = "", 

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

348 correlation_id: str | None = None, 

349 *, 

350 maxlen: int | None = None, 

351 pipeline: Optional["Pipeline[bytes]"] = None, 

352 ) -> bytes: 

353 cmd = RedisPublishCommand( (empty)

354 message, 

355 stream=stream or self.stream.name, 

356 reply_to=reply_to or self.reply_to, 

357 headers=self.headers | (headers or {}), 

358 correlation_id=correlation_id or gen_cor_id(), 

359 maxlen=maxlen or self.stream.maxlen, 

360 pipeline=pipeline, 

361 _publish_type=PublishType.PUBLISH, 

362 message_format=self.config.message_format, 

363 ) 

364 

365 result: bytes = await self._basic_publish( (empty)

366 cmd, 

367 producer=self.producer, 

368 _extra_middlewares=(), 

369 ) 

370 return result (empty)

371 

372 @override (empty)

373 async def _publish( (empty)

374 self, 

375 cmd: Union["PublishCommand", "RedisPublishCommand"], 

376 *, 

377 _extra_middlewares: Iterable["PublisherMiddleware"], 

378 ) -> None: 

379 """This method should be called in subscriber flow only.""" 

380 cmd = RedisPublishCommand.from_cmd(cmd, message_format=self.config.message_format) (empty)

381 

382 cmd.set_destination(stream=self.stream.name) (empty)

383 

384 cmd.add_headers(self.headers, override=False) (empty)

385 cmd.reply_to = cmd.reply_to or self.reply_to (empty)

386 cmd.maxlen = self.stream.maxlen (empty)

387 

388 await self._basic_publish( (empty)

389 cmd, 

390 producer=self.producer, 

391 _extra_middlewares=_extra_middlewares, 

392 ) 

393 

394 @override (empty)

395 async def request( (empty)

396 self, 

397 message: "SendableMessage" = None, 

398 stream: str | None = None, 

399 *, 

400 maxlen: int | None = None, 

401 correlation_id: str | None = None, 

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

403 timeout: float | None = 30.0, 

404 ) -> "RedisChannelMessage": 

405 cmd = RedisPublishCommand( 

406 message, 

407 stream=stream or self.stream.name, 

408 headers=self.headers | (headers or {}), 

409 correlation_id=correlation_id or gen_cor_id(), 

410 maxlen=maxlen or self.stream.maxlen, 

411 timeout=timeout, 

412 _publish_type=PublishType.REQUEST, 

413 message_format=self.config.message_format, 

414 ) 

415 

416 msg: RedisChannelMessage = await self._basic_request( 

417 cmd, 

418 producer=self.producer, 

419 ) 

420 return msg