Skip to content

KafkaAckableMessage

faststream.kafka.message.KafkaAckableMessage #

KafkaAckableMessage(*args: Any, **kwargs: Any)

Bases: KafkaMessage

Source code in faststream/kafka/message.py 
66
67
68
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self.committed = None

committed instance-attribute #

committed = None

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 correlation_id

processed instance-attribute #

processed = False

consumer instance-attribute #

consumer = consumer

ack async #

ack() -> None

Acknowledge the Kafka message.

Source code in faststream/kafka/message.py 
70
71
72
73
74
async def ack(self) -> None:
    """Acknowledge the Kafka message."""
    if not self.committed:
        await self.consumer.commit()
    await super().ack()

nack async #

nack() -> None

Reject the Kafka message.

Source code in faststream/kafka/message.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def nack(self) -> None:
    """Reject the Kafka message."""
    if not self.committed:
        raw_message = (
            self.raw_message[0]
            if isinstance(self.raw_message, tuple)
            else self.raw_message
        )
        topic_partition = AIOKafkaTopicPartition(
            raw_message.topic,
            raw_message.partition,
        )
        self.consumer.seek(
            partition=topic_partition,
            offset=raw_message.offset,
        )
    await super().nack()

set_decoder #

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

clear_cache #

clear_cache() -> None
Source code in faststream/message/message.py 
68
69
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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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)) is None:
        result = self.__decoded_caches[self.__decoder] = await self.__decoder(self)

    return result

reject async #

reject() -> None
Source code in faststream/message/message.py 
112
113
114
async def reject(self) -> None:
    if self.committed is None:
        self.committed = AckStatus.REJECTED