Skip to content

RedisFakePublisher

faststream.redis.publisher.fake.RedisFakePublisher #

RedisFakePublisher(producer, channel, message_format)

Bases: FakePublisher

Publisher Interface implementation to use as RPC or REPLY TO answer publisher.

Source code in faststream/redis/publisher/fake.py 
15
16
17
18
19
20
21
22
23
def __init__(
    self,
    producer: "ProducerProto[RedisPublishCommand]",
    channel: str,
    message_format: type["MessageFormat"],
) -> None:
    super().__init__(producer=producer)
    self.channel = channel
    self.message_format = message_format

channel instance-attribute #

channel = channel

message_format instance-attribute #

message_format = message_format

patch_command #

patch_command(cmd)
Source code in faststream/redis/publisher/fake.py 
25
26
27
28
29
30
31
32
def patch_command(
    self,
    cmd: Union["PublishCommand", "RedisPublishCommand"],
) -> "RedisPublishCommand":
    cmd = super().patch_command(cmd)
    real_cmd = RedisPublishCommand.from_cmd(cmd, message_format=self.message_format)
    real_cmd.destination = self.channel
    return real_cmd

publish async #

publish(message, /, *, correlation_id=None)
Source code in faststream/_internal/endpoint/publisher/fake.py 
49
50
51
52
53
54
55
56
57
58
59
60
async def publish(
    self,
    message: SendableMessage,
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None:
    msg = (
        f"`{self.__class__.__name__}` can be used only to publish "
        "a response for `reply-to` or `RPC` messages."
    )
    raise NotImplementedError(msg)

request async #

request(message, /, *, correlation_id=None)
Source code in faststream/_internal/endpoint/publisher/fake.py 
62
63
64
65
66
67
68
69
70
71
72
73
async def request(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: str | None = None,
) -> Any:
    msg = (
        f"`{self.__class__.__name__}` can be used only to publish "
        "a response for `reply-to` or `RPC` messages."
    )
    raise NotImplementedError(msg)