Skip to content

TopicPartition

faststream.confluent.TopicPartition #

TopicPartition(
    topic: str,
    partition: int = -1,
    offset: int = -1001,
    metadata: str | None = None,
    leader_epoch: int | None = None,
    *,
    declare: bool = True,
)

Initialize the Kafka topic partition.

PARAMETER DESCRIPTION
topic

Kafka topic name.

TYPE: str

partition

Partition number to assign.

TYPE: int DEFAULT: -1

offset

Offset to start consuming from.

TYPE: int DEFAULT: -1001

metadata

Application-specific metadata to attach to the partition.

TYPE: str | None DEFAULT: None

leader_epoch

Leader epoch of the offset.

TYPE: int | None DEFAULT: None

declare

Whether to create the topic automatically or just connect to it. Missing topics are not created and their absence is not reported, so set it to False for topics provisioned by someone else.

TYPE: bool DEFAULT: True

Source code in faststream/confluent/schemas/partition.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    topic: str,
    partition: int = -1,
    offset: int = -1001,
    metadata: str | None = None,
    leader_epoch: int | None = None,
    *,
    declare: bool = True,
) -> None:
    """Initialize the Kafka topic partition.

    Args:
        topic: Kafka topic name.
        partition: Partition number to assign.
        offset: Offset to start consuming from.
        metadata: Application-specific metadata to attach to the partition.
        leader_epoch: Leader epoch of the offset.
        declare: Whether to create the topic automatically or just connect to it.
            Missing topics are not created and their absence is not reported,
            so set it to `False` for topics provisioned by someone else.
    """
    self.topic = topic
    self.partition = partition
    self.offset = offset
    self.metadata = metadata
    self.leader_epoch = leader_epoch
    self.declare = declare

topic instance-attribute #

topic = topic

partition instance-attribute #

partition = partition

offset instance-attribute #

offset = offset

metadata instance-attribute #

metadata = metadata

leader_epoch instance-attribute #

leader_epoch = leader_epoch

declare instance-attribute #

declare = declare

to_confluent #

to_confluent() -> TopicPartition
Source code in faststream/confluent/schemas/partition.py 
55
56
57
58
59
60
61
62
63
64
65
def to_confluent(self) -> ConfluentPartition:
    kwargs: _TopicKwargs = {
        "topic": self.topic,
        "partition": self.partition,
        "offset": self.offset,
    }
    if self.metadata is not None:
        kwargs["metadata"] = self.metadata
    if self.leader_epoch is not None:
        kwargs["leader_epoch"] = self.leader_epoch
    return ConfluentPartition(**kwargs)

add_prefix #

add_prefix(prefix: str) -> TopicPartition
Source code in faststream/confluent/schemas/partition.py 
67
68
69
70
71
72
73
74
75
def add_prefix(self, prefix: str) -> "TopicPartition":
    return TopicPartition(
        topic=f"{prefix}{self.topic}",
        partition=self.partition,
        offset=self.offset,
        metadata=self.metadata,
        leader_epoch=self.leader_epoch,
        declare=self.declare,
    )