Skip to content

DefaultPublisher

faststream.confluent.publisher.usecase.DefaultPublisher #

DefaultPublisher(
    config: KafkaPublisherConfig,
    specifcication: PublisherSpecification[Any, Any],
)

Bases: LogicPublisher

Source code in faststream/confluent/publisher/usecase.py 
82
83
84
85
86
87
88
89
def __init__(
    self,
    config: "KafkaPublisherConfig",
    specifcication: "PublisherSpecification[Any, Any]",
) -> None:
    super().__init__(config, specifcication)

    self.key = config.key

key instance-attribute #

key = key

specification instance-attribute #

specification = specification

middlewares instance-attribute #

middlewares = middlewares

mock instance-attribute #

mock = MagicMock()

partition instance-attribute #

partition = partition

reply_to instance-attribute #

reply_to = reply_to

headers instance-attribute #

headers = headers or {}

topic property #

topic: str

publish async #

publish(
    message: SendableMessage,
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
    no_confirm: Literal[True] = ...,
) -> Future[Message | None]

publish(
    message: SendableMessage,
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
    no_confirm: Literal[False] = False,
) -> Message | None

publish(
    message: SendableMessage,
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
    no_confirm: bool = False,
) -> Future[Message | None] | Message | None

publish(
    message: SendableMessage,
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
    no_confirm: bool = False,
) -> Future[Message | None] | Message | None
Source code in faststream/confluent/publisher/usecase.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@override
async def publish(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    reply_to: str = "",
    no_confirm: bool = False,
) -> asyncio.Future[Message | None] | Message | None:
    cmd = KafkaPublishCommand(
        message,
        topic=topic or self.topic,
        key=key or self.key,
        partition=partition if partition is not None else self.partition,
        reply_to=reply_to or self.reply_to,
        headers=self.headers | (headers or {}),
        correlation_id=correlation_id or gen_cor_id(),
        timestamp_ms=timestamp_ms,
        no_confirm=no_confirm,
        _publish_type=PublishType.PUBLISH,
    )
    msg: asyncio.Future[Message | None] | Message | None = await self._basic_publish(
        cmd,
        producer=self._outer_config.producer,
        _extra_middlewares=(),
    )
    return msg

request async #

request(
    message: SendableMessage,
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    timeout: float = 0.5,
) -> KafkaMessage
Source code in faststream/confluent/publisher/usecase.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@override
async def request(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    timeout: float = 0.5,
) -> "KafkaMessage":
    return await super().request(
        message,
        topic=topic,
        key=key or self.key,
        partition=partition,
        timestamp_ms=timestamp_ms,
        headers=headers,
        correlation_id=correlation_id,
        timeout=timeout,
    )

start async #

start() -> None
Source code in faststream/_internal/endpoint/publisher/usecase.py 
51
52
async def start(self) -> None:
    pass

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()

flush async #

flush() -> None
Source code in faststream/confluent/publisher/usecase.py 
76
77
78
async def flush(self) -> None:
    producer = cast("AsyncConfluentFastProducer", self._outer_config.producer)
    await producer.flush()