Skip to content

AsyncConfluentParser

faststream.confluent.parser.AsyncConfluentParser #

AsyncConfluentParser(is_manual: bool = False)

A class to parse Kafka messages.

Source code in faststream/confluent/parser.py 
32
33
34
def __init__(self, is_manual: bool = False) -> None:
    self.is_manual = is_manual
    self._consumer: ConsumerProtocol = FAKE_CONSUMER

is_manual instance-attribute #

is_manual = is_manual

parse_message async #

parse_message(message: Message) -> KafkaMessage

Parses a Kafka message.

Source code in faststream/confluent/parser.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def parse_message(
    self,
    message: "Message",
) -> KafkaMessage:
    """Parses a Kafka message."""
    headers = _parse_msg_headers(cast("_HeadersInput", message.headers() or ()))

    body = message.value() or b""
    offset = message.offset()
    _, timestamp = message.timestamp()

    return KafkaMessage(
        body=body,
        headers=headers,
        reply_to=headers.get("reply_to", ""),
        content_type=headers.get("content-type"),
        message_id=f"{offset}-{timestamp}",
        correlation_id=headers.get("correlation_id"),
        raw_message=message,
        consumer=self._consumer,
        is_manual=self.is_manual,
    )

parse_batch async #

parse_batch(message: tuple[Message, ...]) -> KafkaMessage

Parses a batch of messages from a Kafka consumer.

Source code in faststream/confluent/parser.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def parse_batch(
    self,
    message: tuple["Message", ...],
) -> KafkaMessage:
    """Parses a batch of messages from a Kafka consumer."""
    body: list[Any] = []
    batch_headers: list[dict[str, str]] = []

    first = message[0]
    last = message[-1]

    for m in message:
        body.append(m.value() or b"")
        batch_headers.append(
            _parse_msg_headers(cast("_HeadersInput", m.headers() or ()))
        )

    headers = next(iter(batch_headers), {})

    _, first_timestamp = first.timestamp()

    return KafkaMessage(
        body=body,
        headers=headers,
        batch_headers=batch_headers,
        reply_to=headers.get("reply_to", ""),
        content_type=headers.get("content-type"),
        message_id=f"{first.offset()}-{last.offset()}-{first_timestamp}",
        correlation_id=headers.get("correlation_id"),
        raw_message=message,
        consumer=self._consumer,
        is_manual=self.is_manual,
    )

decode_message async #

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

Decodes a message.

Source code in faststream/confluent/parser.py 
 96
 97
 98
 99
100
101
async def decode_message(  # ruff: ignore[no-self-use]
    self,
    msg: "StreamMessage[Message]",
) -> "DecodedMessage":
    """Decodes a message."""
    return decode_message(msg)

decode_batch async #

decode_batch(
    msg: StreamMessage[tuple[Message, ...]],
) -> DecodedMessage

Decode a batch of messages.

Source code in faststream/confluent/parser.py 
103
104
105
106
107
108
async def decode_batch(
    self,
    msg: "StreamMessage[tuple[Message, ...]]",
) -> "DecodedMessage":
    """Decode a batch of messages."""
    return [decode_message(await self.parse_message(m)) for m in msg.raw_message]