Skip to content

Topic

faststream.confluent.schemas.topic.Topic #

Topic(
    name: str,
    *,
    num_partitions: int = 1,
    replication_factor: int = 1,
    declare: bool = True,
)

Bases: NameRequired

A Kafka topic with its creation settings.

Can be used instead of a plain topic name to configure the topic FastStream creates for you.

You can find information about all options in the official Kafka documentation:

https://kafka.apache.org/documentation/#topicconfigs

Initialize the Kafka topic.

PARAMETER DESCRIPTION
name

Kafka topic name.

TYPE: str

num_partitions

Number of partitions to create the topic with.

TYPE: int DEFAULT: 1

replication_factor

Replication factor to create the topic with.

TYPE: int DEFAULT: 1

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/topic.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    name: str,
    *,
    num_partitions: int = 1,
    replication_factor: int = 1,
    declare: bool = True,
) -> None:
    """Initialize the Kafka topic.

    Args:
        name: Kafka topic name.
        num_partitions: Number of partitions to create the topic with.
        replication_factor: Replication factor to create the topic with.
        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.
    """
    super().__init__(name)

    self.num_partitions = num_partitions
    self.replication_factor = replication_factor
    self.declare = declare

num_partitions instance-attribute #

num_partitions = num_partitions

replication_factor instance-attribute #

replication_factor = replication_factor

declare instance-attribute #

declare = declare

name instance-attribute #

name = name

add_prefix #

add_prefix(prefix: str) -> Topic
Source code in faststream/confluent/schemas/topic.py 
78
79
80
81
82
83
84
def add_prefix(self, prefix: str) -> "Topic":
    return Topic(
        f"{prefix}{self.name}",
        num_partitions=self.num_partitions,
        replication_factor=self.replication_factor,
        declare=self.declare,
    )

to_confluent #

to_confluent() -> NewTopic
Source code in faststream/confluent/schemas/topic.py 
86
87
88
89
90
91
def to_confluent(self) -> NewTopic:
    return NewTopic(
        self.name,
        num_partitions=self.num_partitions,
        replication_factor=self.replication_factor,
    )

validate classmethod #

validate(value: str | Self, **kwargs: Any) -> Self

validate(value: None, **kwargs: Any) -> None

validate(
    value: str | Self | None, **kwargs: Any
) -> Self | None

Factory to create object.

Source code in faststream/_internal/proto.py 
30
31
32
33
34
35
@classmethod
def validate(cls, value: str | Self | None, **kwargs: Any) -> Self | None:
    """Factory to create object."""
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value