Skip to content

LogicPublisher

faststream.redis.publisher.usecase.LogicPublisher #

LogicPublisher(
    config: RedisPublisherConfig,
    specification: PublisherSpecification[Any, Any],
)

Bases: PublisherUsecase

A class to represent a Redis publisher.

Source code in faststream/redis/publisher/usecase.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    config: "RedisPublisherConfig",
    specification: "PublisherSpecification[Any, Any]",
) -> None:
    super().__init__(config, specification)

    self.config = config

    self.reply_to = config.reply_to
    self.headers = config.headers or {}

    self.producer = self.config._outer_config.producer

config instance-attribute #

config = config

reply_to instance-attribute #

reply_to = reply_to

headers instance-attribute #

headers = headers or {}

producer instance-attribute #

producer = producer

specification instance-attribute #

specification = specification

middlewares instance-attribute #

middlewares = middlewares

mock instance-attribute #

mock = MagicMock()

start async #

start() -> None
Source code in faststream/redis/publisher/usecase.py 
46
47
48
49
50
51
52
53
54
55
56
57
async def start(self) -> None:
    await super().start()

    broker_producer = self.config._outer_config.producer

    self.producer = RedisFastProducer(
        connection=self.config._outer_config.connection,
        parser=broker_producer._parser.custom_func,
        decoder=broker_producer._decoder.custom_func,
        message_format=self.config.message_format,
        serializer=self.config._outer_config.fd_config._serializer,
    )

subscriber_property abstractmethod #

subscriber_property(*, name_only: bool) -> dict[str, Any]
Source code in faststream/redis/publisher/usecase.py 
59
60
61
@abstractmethod
def subscriber_property(self, *, name_only: bool) -> dict[str, Any]:
    raise NotImplementedError

publish abstractmethod async #

publish(
    message: SendableMessage,
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None

Public method to publish a message.

Should be called by user only broker.publisher(...).publish(...).

Source code in faststream/_internal/endpoint/publisher/proto.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
@abstractmethod
async def publish(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None:
    """Public method to publish a message.

    Should be called by user only `broker.publisher(...).publish(...)`.
    """
    ...

request abstractmethod async #

request(
    message: SendableMessage,
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None

Publishes a message synchronously.

Source code in faststream/_internal/endpoint/publisher/proto.py 
44
45
46
47
48
49
50
51
52
53
@abstractmethod
async def request(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None:
    """Publishes a message synchronously."""
    ...

set_test #

set_test(*, mock: MagicMock, with_fake: bool) -> None

Turn publisher to testing mode.

Source code in faststream/_internal/endpoint/publisher/usecase.py 
54
55
56
57
58
59
60
61
62
def set_test(
    self,
    *,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode."""
    self.mock = mock
    self._fake_handler = with_fake

reset_test #

reset_test() -> None

Turn off publisher's testing mode.

Source code in faststream/_internal/endpoint/publisher/usecase.py 
64
65
66
67
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock.reset_mock()

schema #

schema() -> dict[str, PublisherSpec]
Source code in faststream/_internal/endpoint/publisher/usecase.py 
146
147
def schema(self) -> dict[str, "PublisherSpec"]:
    return self.specification.get_schema()