Skip to content

LogicPublisher

faststream.confluent.publisher.usecase.LogicPublisher #

LogicPublisher(*, topic, partition, headers, reply_to, broker_middlewares, middlewares, schema_, title_, description_, include_in_schema)

Bases: PublisherUsecase[MsgType]

A class to publish messages to a Kafka topic.

Source code in faststream/confluent/publisher/usecase.py 
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
66
67
68
69
70
def __init__(
    self,
    *,
    topic: str,
    partition: Optional[int],
    headers: Optional[Dict[str, str]],
    reply_to: Optional[str],
    # Publisher args
    broker_middlewares: Sequence["BrokerMiddleware[MsgType]"],
    middlewares: Sequence["PublisherMiddleware"],
    # AsyncAPI args
    schema_: Optional[Any],
    title_: Optional[str],
    description_: Optional[str],
    include_in_schema: bool,
) -> None:
    super().__init__(
        broker_middlewares=broker_middlewares,
        middlewares=middlewares,
        # AsyncAPI args
        schema_=schema_,
        title_=title_,
        description_=description_,
        include_in_schema=include_in_schema,
    )

    self.topic = topic
    self.partition = partition
    self.reply_to = reply_to
    self.headers = headers

    self._producer = None

title_ instance-attribute #

title_ = title_

description_ instance-attribute #

description_ = description_

include_in_schema instance-attribute #

include_in_schema = include_in_schema

name property #

name

Returns the name of the API operation.

description property #

description

Returns the description of the API operation.

schema_ instance-attribute #

schema_ = schema_

mock instance-attribute #

mock = None

calls instance-attribute #

calls = []

topic instance-attribute #

topic = topic

partition instance-attribute #

partition = partition

reply_to instance-attribute #

reply_to = reply_to

headers instance-attribute #

headers = headers

publish abstractmethod async #

publish(message, /, *, correlation_id=None, _extra_middlewares=())

Publishes a message asynchronously.

Source code in faststream/broker/publisher/proto.py 
57
58
59
60
61
62
63
64
65
66
67
@abstractmethod
async def publish(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: Optional[str] = None,
    _extra_middlewares: Sequence["PublisherMiddleware"] = (),
) -> Optional[Any]:
    """Publishes a message asynchronously."""
    ...

setup #

setup(*, producer)
Source code in faststream/broker/publisher/usecase.py 
 97
 98
 99
100
101
102
103
@override
def setup(  # type: ignore[override]
    self,
    *,
    producer: Optional["ProducerProto"],
) -> None:
    self._producer = producer

schema #

schema()

Returns the schema of the API operation as a dictionary of channel names and channel objects.

Source code in faststream/asyncapi/abc.py 
30
31
32
33
34
35
def schema(self) -> Dict[str, Channel]:
    """Returns the schema of the API operation as a dictionary of channel names and channel objects."""
    if self.include_in_schema:
        return self.get_schema()
    else:
        return {}

add_middleware #

add_middleware(middleware)
Source code in faststream/broker/publisher/usecase.py 
94
95
def add_middleware(self, middleware: "BrokerMiddleware[MsgType]") -> None:
    self._broker_middlewares = (*self._broker_middlewares, middleware)

create abstractmethod staticmethod #

create()

Abstract factory to create a real Publisher.

Source code in faststream/broker/publisher/proto.py 
 97
 98
 99
100
101
@staticmethod
@abstractmethod
def create() -> "PublisherProto[MsgType]":
    """Abstract factory to create a real Publisher."""
    ...

get_name abstractmethod #

get_name()

Name property fallback.

Source code in faststream/asyncapi/abc.py 
16
17
18
19
@abstractmethod
def get_name(self) -> str:
    """Name property fallback."""
    raise NotImplementedError()

get_description #

get_description()

Description property fallback.

Source code in faststream/asyncapi/abc.py 
26
27
28
def get_description(self) -> Optional[str]:
    """Description property fallback."""
    return None

get_schema abstractmethod #

get_schema()

Generate AsyncAPI schema.

Source code in faststream/asyncapi/abc.py 
37
38
39
40
@abstractmethod
def get_schema(self) -> Dict[str, Channel]:
    """Generate AsyncAPI schema."""
    raise NotImplementedError()

get_payloads #

get_payloads()
Source code in faststream/broker/publisher/usecase.py 
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
168
169
170
171
172
173
174
def get_payloads(self) -> List[Tuple["AnyDict", str]]:
    payloads: List[Tuple[AnyDict, str]] = []

    if self.schema_:
        params = {"response__": (self.schema_, ...)}

        call_model: CallModel[Any, Any] = CallModel(
            call=lambda: None,
            model=create_model("Fake"),
            response_model=create_model(  # type: ignore[call-overload]
                "",
                __config__=get_config_base(),
                **params,
            ),
            params=params,
        )

        body = get_response_schema(
            call_model,
            prefix=f"{self.name}:Message",
        )
        if body:  # pragma: no branch
            payloads.append((body, ""))

    else:
        for call in self.calls:
            call_model = build_call_model(call)
            body = get_response_schema(
                call_model,
                prefix=f"{self.name}:Message",
            )
            if body:
                payloads.append((body, to_camelcase(unwrap(call).__name__)))

    return payloads

set_test #

set_test(*, mock, with_fake)

Turn publisher to testing mode.

Source code in faststream/broker/publisher/usecase.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def set_test(
    self,
    *,
    mock: Annotated[
        MagicMock,
        Doc("Mock object to check in tests."),
    ],
    with_fake: Annotated[
        bool,
        Doc("Whetevet publisher's fake subscriber created or not."),
    ],
) -> None:
    """Turn publisher to testing mode."""
    self.mock = mock
    self._fake_handler = with_fake

reset_test #

reset_test()

Turn off publisher's testing mode.

Source code in faststream/broker/publisher/usecase.py 
121
122
123
124
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock = None

add_prefix #

add_prefix(prefix)
Source code in faststream/confluent/publisher/usecase.py 
75
76
def add_prefix(self, prefix: str) -> None:
    self.topic = "".join((prefix, self.topic))

request async #

request(message, topic='', *, key=None, partition=None, timestamp_ms=None, headers=None, correlation_id=None, timeout=0.5, _extra_middlewares=())
Source code in faststream/confluent/publisher/usecase.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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
@override
async def request(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: Optional[bytes] = None,
    partition: Optional[int] = None,
    timestamp_ms: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None,
    correlation_id: Optional[str] = None,
    timeout: float = 0.5,
    # publisher specific
    _extra_middlewares: Iterable["PublisherMiddleware"] = (),
) -> "KafkaMessage":
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    kwargs: AnyDict = {
        "key": key,
        # basic args
        "timeout": timeout,
        "timestamp_ms": timestamp_ms,
        "topic": topic or self.topic,
        "partition": partition or self.partition,
        "headers": headers or self.headers,
        "correlation_id": correlation_id or gen_cor_id(),
    }

    request: AsyncFunc = self._producer.request

    for pub_m in chain(
        self._middlewares[::-1],
        (
            _extra_middlewares
            or (m(None).publish_scope for m in self._broker_middlewares[::-1])
        ),
    ):
        request = partial(pub_m, request)

    published_msg = await request(message, **kwargs)

    async with AsyncExitStack() as stack:
        return_msg: Callable[[KafkaMessage], Awaitable[KafkaMessage]] = return_input
        for m in self._broker_middlewares[::-1]:
            mid = m(published_msg)
            await stack.enter_async_context(mid)
            return_msg = partial(mid.consume_scope, return_msg)

        parsed_msg = await self._producer._parser(published_msg)
        parsed_msg._decoded_body = await self._producer._decoder(parsed_msg)
        parsed_msg._source_type = SourceType.Response
        return await return_msg(parsed_msg)

    raise AssertionError("unreachable")