Skip to content

AsyncConfluentFastProducer

faststream.confluent.publisher.producer.AsyncConfluentFastProducer #

AsyncConfluentFastProducer(producer, parser, decoder)

Bases: ProducerProto

A class to represent Kafka producer.

Source code in faststream/confluent/publisher/producer.py 
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(
    self,
    producer: "AsyncConfluentProducer",
    parser: Optional["CustomCallable"],
    decoder: Optional["CustomCallable"],
) -> None:
    self._producer = producer

    # NOTE: register default parser to be compatible with request
    default = AsyncConfluentParser
    self._parser = resolve_custom_func(parser, default.parse_message)
    self._decoder = resolve_custom_func(decoder, default.decode_message)

publish async #

publish(message, topic, *, key=None, partition=None, timestamp_ms=None, headers=None, correlation_id='', reply_to='', no_confirm=False)

Publish a message to a topic.

Source code in faststream/confluent/publisher/producer.py 
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
66
67
68
69
70
@override
async def publish(  # type: ignore[override]
    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: str = "",
    reply_to: str = "",
    no_confirm: bool = False,
) -> None:
    """Publish a message to a topic."""
    message, content_type = encode_message(message)

    headers_to_send = {
        "content-type": content_type or "",
        "correlation_id": correlation_id,
        **(headers or {}),
    }

    if reply_to:
        headers_to_send["reply_to"] = headers_to_send.get(
            "reply_to",
            reply_to,
        )

    await self._producer.send(
        topic=topic,
        value=message,
        key=key,
        partition=partition,
        timestamp_ms=timestamp_ms,
        headers=[(i, (j or "").encode()) for i, j in headers_to_send.items()],
        no_confirm=no_confirm,
    )

stop async #

stop()
Source code in faststream/confluent/publisher/producer.py 
72
73
async def stop(self) -> None:
    await self._producer.stop()

publish_batch async #

publish_batch(*msgs, topic, partition=None, timestamp_ms=None, headers=None, reply_to='', correlation_id='', no_confirm=False)

Publish a batch of messages to a topic.

Source code in faststream/confluent/publisher/producer.py 
 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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
async def publish_batch(
    self,
    *msgs: "SendableMessage",
    topic: str,
    partition: Optional[int] = None,
    timestamp_ms: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None,
    reply_to: str = "",
    correlation_id: str = "",
    no_confirm: bool = False,
) -> None:
    """Publish a batch of messages to a topic."""
    batch = self._producer.create_batch()

    headers_to_send = {"correlation_id": correlation_id, **(headers or {})}

    if reply_to:
        headers_to_send["reply_to"] = headers_to_send.get(
            "reply_to",
            reply_to,
        )

    for msg in msgs:
        message, content_type = encode_message(msg)

        if content_type:
            final_headers = {
                "content-type": content_type,
                **headers_to_send,
            }
        else:
            final_headers = headers_to_send.copy()

        batch.append(
            key=None,
            value=message,
            timestamp=timestamp_ms,
            headers=[(i, j.encode()) for i, j in final_headers.items()],
        )

    await self._producer.send_batch(
        batch,
        topic,
        partition=partition,
        no_confirm=no_confirm,
    )

request async #

request(*args, **kwargs)
Source code in faststream/confluent/publisher/producer.py 
122
123
124
125
126
@override
async def request(self, *args: Any, **kwargs: Any) -> Optional[Any]:
    raise OperationForbiddenError(
        "Kafka doesn't support `request` method without test client."
    )