Skip to content

AioKafkaParser

faststream.kafka.parser.AioKafkaParser #

AioKafkaParser(
    msg_class: type[KafkaMessage],
    regex: Optional[Pattern[str]],
)

A class to parse Kafka messages.

Source code in faststream/kafka/parser.py 
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    msg_class: type[KafkaMessage],
    regex: Optional["Pattern[str]"],
) -> None:
    self.msg_class = msg_class
    self.regex = regex

    self._consumer: ConsumerProtocol = FAKE_CONSUMER

msg_class instance-attribute #

msg_class = msg_class

regex instance-attribute #

regex = regex

parse_message async #

parse_message(
    message: Union[ConsumerRecord, KafkaRawMessage],
) -> StreamMessage[ConsumerRecord]

Parses a Kafka message.

Source code in faststream/kafka/parser.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
async def parse_message(
    self,
    message: Union["ConsumerRecord", "KafkaRawMessage"],
) -> "StreamMessage[ConsumerRecord]":
    """Parses a Kafka message."""
    headers = {
        key: value.decode(errors="replace")
        for key, value in message.headers
        if value is not None
    }

    return self.msg_class(
        body=message.value or b"",
        headers=headers,
        reply_to=headers.get("reply_to", ""),
        content_type=headers.get("content-type"),
        message_id=f"{message.offset}-{message.timestamp}",
        correlation_id=headers.get("correlation_id"),
        raw_message=message,
        path=match_path(self.regex, message.topic),
        consumer=getattr(message, "consumer", self._consumer),
    )

decode_message async #

decode_message(
    msg: StreamMessage[ConsumerRecord],
) -> DecodedMessage

Decodes a message.

Source code in faststream/kafka/parser.py 
66
67
68
69
70
71
async def decode_message(  # ruff: ignore[no-self-use]
    self,
    msg: "StreamMessage[ConsumerRecord]",
) -> "DecodedMessage":
    """Decodes a message."""
    return decode_message(msg)