Skip to content

KafkaMessage

faststream.confluent.message.KafkaMessage #

KafkaMessage(
    *args: Any,
    consumer: ConsumerProtocol,
    is_manual: bool = False,
    **kwargs: Any,
)

Bases: StreamMessage[Message | tuple[Message, ...]]

Represents a Kafka message in the FastStream framework.

This class extends StreamMessage and is specialized for handling confluent_kafka.Message objects.

Source code in faststream/confluent/message.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    *args: Any,
    consumer: ConsumerProtocol,
    is_manual: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(*args, **kwargs)

    self.consumer = consumer

    self.is_manual = is_manual
    if not is_manual:
        self.committed = AckStatus.ACKED

consumer instance-attribute #

consumer = consumer

is_manual instance-attribute #

is_manual = is_manual

committed instance-attribute #

committed = AckStatus.ACKED

raw_headers property #

raw_headers: dict[str, bytes | None]

Original header bytes, or the first message's headers for a batch.

batch_raw_headers property #

batch_raw_headers: list[dict[str, bytes | None]]

Original header bytes for every batch message; empty for a single message.

raw_message instance-attribute #

raw_message = raw_message

body instance-attribute #

body = body

reply_to instance-attribute #

reply_to = reply_to

content_type instance-attribute #

content_type = content_type

source_type instance-attribute #

source_type = source_type

headers instance-attribute #

headers = headers or {}

batch_headers instance-attribute #

batch_headers = batch_headers or []

path instance-attribute #

path = path or {}

correlation_id instance-attribute #

correlation_id = correlation_id or str(uuid4())

message_id instance-attribute #

message_id = message_id or self.correlation_id

processed instance-attribute #

processed = False

ack async #

ack() -> None

Acknowledge the Kafka message.

Source code in faststream/confluent/message.py 
79
80
81
82
83
async def ack(self) -> None:
    """Acknowledge the Kafka message."""
    if self.is_manual and not self.committed:
        await self.consumer.commit()
    await super().ack()

reject async #

reject() -> None
Source code in faststream/confluent/message.py 
85
86
async def reject(self) -> None:
    await self.ack()

nack async #

nack() -> None

Reject the Kafka message.

Source code in faststream/confluent/message.py 
 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
async def nack(self) -> None:
    """Reject the Kafka message."""
    if self.is_manual and not self.committed:
        # A batch may span several partitions: rewind each one to its earliest
        # record, so none of them is committed past an unprocessed message.
        records: tuple[Message, ...] = (
            self.raw_message
            if isinstance(self.raw_message, tuple)
            else (self.raw_message,)
        )
        earliest: dict[tuple[str, int], int] = {}
        for record in records:
            topic = record.topic()
            partition = record.partition()
            offset = record.offset()
            if topic is None or partition is None or offset is None:
                continue
            known = earliest.get((topic, partition))
            if known is None or offset < known:
                earliest[topic, partition] = offset
        for (topic, partition), offset in earliest.items():
            await self.consumer.seek(
                topic=topic,
                partition=partition,
                offset=offset,
            )
    await super().nack()

set_decoder #

set_decoder(decoder: AsyncCallable) -> None
Source code in faststream/message/message.py 
69
70
def set_decoder(self, decoder: "AsyncCallable") -> None:
    self.__decoder = decoder

clear_cache #

clear_cache() -> None
Source code in faststream/message/message.py 
72
73
def clear_cache(self) -> None:
    self.__decoded_caches.clear()

decode async #

decode() -> Optional[Any]

Serialize the message by lazy decoder.

Returns a cache after first usage. To prevent such behavior, please call message.clear_cache() after message.body changes.

Source code in faststream/message/message.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def decode(self) -> Optional["Any"]:
    """Serialize the message by lazy decoder.

    Returns a cache after first usage. To prevent such behavior, please call
    `message.clear_cache()` after `message.body` changes.
    """
    assert self.__decoder, "You should call `set_decoder()` method first."

    if (
        result := self.__decoded_caches.get(self.__decoder, _NOT_CACHED)
    ) is _NOT_CACHED:
        result = self.__decoded_caches[self.__decoder] = await self.__decoder(self)

    return result