Skip to content

FakeProducer

faststream.redis.testing.FakeProducer #

FakeProducer(
    broker: RedisBroker,
    brokers: Sequence[RedisBroker],
    config: ParserConfig,
)

Bases: RedisFastProducer

Source code in faststream/redis/testing.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def __init__(
    self,
    broker: RedisBroker,
    brokers: Sequence[RedisBroker],
    config: ParserConfig,
) -> None:
    self.broker = broker
    self.brokers = brokers
    self._fake_config = config

    default = RedisPubSubParser(config)

    self._parser = ParserComposition(
        broker._parser,
        default.parse_message,
    )
    self._decoder = ParserComposition(
        broker._decoder,
        default.decode_message,
    )
    self.codec = broker.config.broker_codec or DefaultCodec()

broker instance-attribute #

broker = broker

brokers instance-attribute #

brokers = brokers

codec instance-attribute #

codec = broker_codec or DefaultCodec()

subscribers property #

subscribers: Iterable[LogicSubscriber]

serializer instance-attribute #

serializer = serializer

publish async #

publish(cmd: RedisPublishCommand) -> int | bytes
Source code in faststream/redis/testing.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@override
async def publish(self, cmd: "RedisPublishCommand") -> int | bytes:
    body = await build_message(
        message=cmd.body,
        reply_to=cmd.reply_to,
        correlation_id=cmd.correlation_id or gen_cor_id(),
        headers=cmd.headers,
        message_format=cmd.message_format,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
    )

    destination = _make_destination_kwargs(cmd)
    visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())

    for handler in self.subscribers:  # pragma: no branch
        for visitor in visitors:
            if visited_ch := visitor.visit(**destination, sub=handler):
                msg = visitor.get_message(
                    visited_ch,
                    body,
                    handler,  # type: ignore[arg-type]
                )

                await self._execute_handler(msg, handler)

    return 0

request async #

request(cmd: RedisPublishCommand) -> PubSubMessage
Source code in faststream/redis/testing.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@override
async def request(self, cmd: "RedisPublishCommand") -> "PubSubMessage":
    body = await build_message(
        message=cmd.body,
        correlation_id=cmd.correlation_id or gen_cor_id(),
        headers=cmd.headers,
        message_format=cmd.message_format,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
    )

    destination = _make_destination_kwargs(cmd)
    visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())

    for handler in self.subscribers:  # pragma: no branch
        for visitor in visitors:
            if visited_ch := visitor.visit(**destination, sub=handler):
                msg = visitor.get_message(
                    visited_ch,
                    body,
                    handler,  # type: ignore[arg-type]
                )

                with anyio.fail_after(cmd.timeout):
                    return await self._execute_handler(msg, handler)

    raise SubscriberNotFound

publish_batch async #

publish_batch(cmd: RedisPublishCommand) -> int
Source code in faststream/redis/testing.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@override
async def publish_batch(self, cmd: "RedisPublishCommand") -> int:
    data_to_send = [
        await build_message(
            m,
            correlation_id=cmd.correlation_id or gen_cor_id(),
            headers=cmd.headers,
            message_format=cmd.message_format,
            serializer=self.broker.config.fd_config._serializer,
            codec=self.codec,
        )
        for m in cmd.batch_bodies
    ]

    visitor = ListVisitor()
    for handler in self.subscribers:  # pragma: no branch
        if visitor.visit(list=cmd.destination, sub=handler):
            casted_handler = cast("_ListHandlerMixin", handler)

            if casted_handler.list_sub.batch:
                msg = visitor.get_message(
                    channel=cmd.destination,
                    body=data_to_send,
                    sub=casted_handler,
                )

                await self._execute_handler(msg, handler)

    return 0

connect #

connect(
    serializer: Optional[SerializerProto] = None,
    codec: Optional[CodecProto] = None,
) -> None
Source code in faststream/redis/publisher/producer.py 
75
76
77
78
79
80
81
82
def connect(
    self,
    serializer: Optional["SerializerProto"] = None,
    codec: Optional["CodecProto"] = None,
) -> None:
    self.serializer = serializer
    if codec is not None:
        self.codec = codec