Skip to content

BatchBuilder

faststream.confluent.helpers.client.BatchBuilder #

BatchBuilder()

A helper class to build a batch of messages to send to Kafka.

Initializes a new BatchBuilder instance.

Source code in faststream/confluent/helpers/client.py 
368
369
370
def __init__(self) -> None:
    """Initializes a new BatchBuilder instance."""
    self._builder: list[dict[str, Any]] = []

append #

append(
    *,
    value: bytes | str | None = None,
    key: bytes | str | None = None,
    timestamp: int | None = None,
    headers: list[tuple[str, bytes]] | None = None,
) -> None

Appends a message to the batch with optional timestamp, key, value, and headers.

Source code in faststream/confluent/helpers/client.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def append(
    self,
    *,
    value: bytes | str | None = None,
    key: bytes | str | None = None,
    timestamp: int | None = None,
    headers: list[tuple[str, bytes]] | None = None,
) -> None:
    """Appends a message to the batch with optional timestamp, key, value, and headers."""
    if key is None and value is None:
        raise KafkaException(
            KafkaError(40, reason="Both key and value can't be None"),
        )

    self._builder.append(
        {
            "timestamp_ms": timestamp or round(time() * 1000),
            "key": key,
            "value": value,
            "headers": headers or [],
        },
    )