Skip to content

AioKafkaBatchParser

faststream.kafka.parser.AioKafkaBatchParser #

AioKafkaBatchParser(msg_class, regex)

Bases: AioKafkaParser

Source code in faststream/kafka/parser.py 
20
21
22
23
24
25
26
def __init__(
    self,
    msg_class: Type[KafkaMessage],
    regex: Optional["Pattern[str]"],
) -> None:
    self.msg_class = msg_class
    self.regex = regex

msg_class instance-attribute #

msg_class = msg_class

regex instance-attribute #

regex = regex

get_path #

get_path(topic)
Source code in faststream/kafka/parser.py 
55
56
57
58
59
def get_path(self, topic: str) -> Dict[str, str]:
    if self.regex and (match := self.regex.match(topic)):
        return match.groupdict()
    else:
        return {}

parse_message async #

parse_message(message)

Parses a batch of messages from a Kafka consumer.

Source code in faststream/kafka/parser.py 
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
async def parse_message(
    self,
    message: Tuple["ConsumerRecord", ...],
) -> "StreamMessage[Tuple[ConsumerRecord, ...]]":
    """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)
        batch_headers.append({i: j.decode() for i, j in m.headers})

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

    handler: Optional[LogicSubscriber[Any]] = context.get_local("handler_")

    return self.msg_class(
        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", gen_cor_id()),
        raw_message=message,
        path=self.get_path(first.topic),
        consumer=getattr(handler, "consumer", None) or FAKE_CONSUMER,
    )

decode_message async #

decode_message(msg)

Decode a batch of messages.

Source code in faststream/kafka/parser.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def decode_message(
    self,
    msg: "StreamMessage[Tuple[ConsumerRecord, ...]]",
) -> "DecodedMessage":
    """Decode a batch of messages."""
    # super() should be here due python can't find it in comprehension
    super_obj = cast(AioKafkaParser, super())

    return [
        decode_message(await super_obj.parse_message(m)) for m in msg.raw_message
    ]