Skip to content

LogicPublisher

faststream.confluent.publisher.usecase.LogicPublisher #

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

Bases: PublisherUsecase

A class to publish messages to a Kafka topic.

Source code in faststream/confluent/publisher/usecase.py 
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    config: "KafkaPublisherConfig",
    specifcication: "PublisherSpecification[Any, Any]",
) -> None:
    super().__init__(config, specifcication)

    self._topic = config.topic
    self.partition = config.partition
    self.reply_to = config.reply_to
    self.headers = config.headers or {}

partition instance-attribute #

partition = config.partition

reply_to instance-attribute #

reply_to = config.reply_to

headers instance-attribute #

headers = config.headers or {}

topic property #

topic: str

is_test instance-attribute #

is_test = False

mock property #

mock: MagicMock

The mock recording the endpoint's calls, available under a test broker.

specification instance-attribute #

specification = specification

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 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@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":
    cmd = KafkaPublishCommand(
        message,
        topic=topic or self.topic,
        key=key,
        partition=partition if partition is not None else self.partition,
        headers=self.headers | (headers or {}),
        correlation_id=correlation_id or self._outer_config.id_generator(),
        timestamp_ms=timestamp_ms,
        timeout=timeout,
        _publish_type=PublishType.REQUEST,
    )

    msg: KafkaMessage = await self._basic_request(
        cmd,
        producer=self._outer_config.producer,
    )
    return msg

flush async #

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

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(...)`.
    """
    ...

assert_called_once_with async #

assert_called_once_with(
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None

Assert the endpoint was called once, with the message described here.

PARAMETER DESCRIPTION
body

The body as a dict, a model or a matcher; it goes through the codec.

TYPE: Any DEFAULT: EMPTY

headers

Headers the message must carry; the rest may carry more.

TYPE: Any DEFAULT: EMPTY

correlation_id

The exact correlation id.

TYPE: Any DEFAULT: EMPTY

reply_to

The exact reply-to destination.

TYPE: Any DEFAULT: EMPTY

content_type

The exact content type.

TYPE: Any DEFAULT: EMPTY

path

The exact path parameters the subject template matched.

TYPE: Any DEFAULT: EMPTY

context

Context paths, as given to Context(), mapped to their values.

TYPE: Mapping[str, Any] DEFAULT: EMPTY

Source code in faststream/_internal/testing/calls.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def assert_called_once_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the endpoint was called once, with the message described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    recorder.mock.assert_called_once()
    await recorder.assert_last_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

assert_called_with async #

assert_called_with(
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None

Assert the last message the endpoint saw is the one described here.

PARAMETER DESCRIPTION
body

The body as a dict, a model or a matcher; it goes through the codec.

TYPE: Any DEFAULT: EMPTY

headers

Headers the message must carry; the rest may carry more.

TYPE: Any DEFAULT: EMPTY

correlation_id

The exact correlation id.

TYPE: Any DEFAULT: EMPTY

reply_to

The exact reply-to destination.

TYPE: Any DEFAULT: EMPTY

content_type

The exact content type.

TYPE: Any DEFAULT: EMPTY

path

The exact path parameters the subject template matched.

TYPE: Any DEFAULT: EMPTY

context

Context paths, as given to Context(), mapped to their values.

TYPE: Mapping[str, Any] DEFAULT: EMPTY

Source code in faststream/_internal/testing/calls.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def assert_called_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the last message the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    await recorder.assert_last_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

assert_any_call async #

assert_any_call(
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None

Assert one of the messages the endpoint saw is the one described here.

PARAMETER DESCRIPTION
body

The body as a dict, a model or a matcher; it goes through the codec.

TYPE: Any DEFAULT: EMPTY

headers

Headers the message must carry; the rest may carry more.

TYPE: Any DEFAULT: EMPTY

correlation_id

The exact correlation id.

TYPE: Any DEFAULT: EMPTY

reply_to

The exact reply-to destination.

TYPE: Any DEFAULT: EMPTY

content_type

The exact content type.

TYPE: Any DEFAULT: EMPTY

path

The exact path parameters the subject template matched.

TYPE: Any DEFAULT: EMPTY

context

Context paths, as given to Context(), mapped to their values.

TYPE: Mapping[str, Any] DEFAULT: EMPTY

Source code in faststream/_internal/testing/calls.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def assert_any_call(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert one of the messages the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    await recorder.assert_any_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

start async #

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

set_test #

set_test(
    *, recorder: CallRecorder | None = None, with_fake: bool
) -> None

Turn publisher to testing mode, sharing recorder when one is given.

Source code in faststream/_internal/endpoint/publisher/usecase.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def set_test(
    self,
    *,
    recorder: CallRecorder | None = None,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode, sharing `recorder` when one is given."""
    self.is_test = True
    if recorder is None:
        self._recorder.reset()
    else:
        self._recorder = recorder
    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 
67
68
69
70
71
72
73
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self.is_test = False
    self._recorder.reset()
    # A shared recorder goes back to the handler it belongs to
    self._recorder = CallRecorder(self.specification.name, self._outer_config)
    self._fake_handler = False

schema #

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