Skip to content

KafkaBrokerConfig

faststream.kafka.configs.KafkaBrokerConfig dataclass #

KafkaBrokerConfig(
    *,
    prefix: str = "",
    include_in_schema: bool | None = True,
    broker_middlewares: Sequence[
        BrokerMiddleware[Any]
    ] = (),
    broker_parser: Optional[CustomCallable] = None,
    broker_decoder: Optional[CustomCallable] = None,
    broker_codec: Optional[CodecProto] = None,
    producer: AioKafkaFastProducer = FakeAioKafkaFastProducer(),
    logger: LoggerState = LoggerState(),
    fd_config: FastDependsConfig = FastDependsConfig(),
    id_generator: IdGenerator = gen_cor_id,
    broker_dependencies: Sequence[Dependant] = (),
    graceful_timeout: float | None = 15.0,
    ack_policy: AckPolicy = (lambda: EMPTY)(),
    extra_context: dict[str, Any] = dict(),
    builder: Callable[..., AIOKafkaConsumer] = lambda: None,
    client_id: str | None = SERVICE_NAME,
    client_rack: str | None = None,
    consumer_only: bool = False,
    _admin_client: Optional[AIOKafkaAdminClient] = None,
)

Bases: BrokerConfig

prefix class-attribute instance-attribute #

prefix: str = ''

include_in_schema class-attribute instance-attribute #

include_in_schema: bool | None = True

broker_middlewares class-attribute instance-attribute #

broker_middlewares: Sequence[BrokerMiddleware[Any]] = ()

broker_parser class-attribute instance-attribute #

broker_parser: Optional[CustomCallable] = None

broker_decoder class-attribute instance-attribute #

broker_decoder: Optional[CustomCallable] = None

broker_codec class-attribute instance-attribute #

broker_codec: Optional[CodecProto] = None

logger class-attribute instance-attribute #

logger: LoggerState = field(default_factory=LoggerState)

fd_config class-attribute instance-attribute #

fd_config: FastDependsConfig = field(
    default_factory=FastDependsConfig
)

id_generator class-attribute instance-attribute #

id_generator: IdGenerator = gen_cor_id

broker_dependencies class-attribute instance-attribute #

broker_dependencies: Sequence[Dependant] = ()

graceful_timeout class-attribute instance-attribute #

graceful_timeout: float | None = 15.0

ack_policy class-attribute instance-attribute #

ack_policy: AckPolicy = field(default_factory=lambda: EMPTY)

extra_context class-attribute instance-attribute #

extra_context: dict[str, Any] = field(default_factory=dict)

context property #

context: ContextRepo

producer class-attribute instance-attribute #

producer: AioKafkaFastProducer = field(
    default_factory=FakeAioKafkaFastProducer
)

builder class-attribute instance-attribute #

builder: Callable[..., AIOKafkaConsumer] = lambda: None

client_id class-attribute instance-attribute #

client_id: str | None = SERVICE_NAME

client_rack class-attribute instance-attribute #

client_rack: str | None = None

consumer_only class-attribute instance-attribute #

consumer_only: bool = False

admin_client property #

admin_client: AIOKafkaAdminClient

add_middleware #

add_middleware(middleware: BrokerMiddleware[Any]) -> None
Source code in faststream/_internal/configs/broker.py 
63
64
def add_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self.broker_middlewares = (*self.broker_middlewares, middleware)

insert_middleware #

insert_middleware(
    middleware: BrokerMiddleware[Any],
) -> None
Source code in faststream/_internal/configs/broker.py 
66
67
def insert_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self.broker_middlewares = (middleware, *self.broker_middlewares)

connect async #

connect(**connection_kwargs: Any) -> None
Source code in faststream/kafka/configs/broker.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def connect(self, **connection_kwargs: Any) -> "None":
    # In consumer-only mode the broker neither produces messages nor needs
    # admin permissions, so skip creating those clients to allow callers
    # to use credentials scoped to read-only ACLs.
    if not self.consumer_only:
        producer = aiokafka.AIOKafkaProducer(**connection_kwargs)
        await self.producer.connect(
            producer,
            serializer=self.fd_config._serializer,
            codec=self.broker_codec or DefaultCodec(),
        )

        admin_options, _ = filter_by_dict(
            AdminClientConnectionParams,
            connection_kwargs,
        )

        self._admin_client = aiokafka.admin.client.AIOKafkaAdminClient(
            **admin_options
        )
        await self._admin_client.start()

    consumer_options, _ = filter_by_dict(
        ConsumerConnectionParams,
        connection_kwargs,
    )
    # client_rack is consumer-only, so it is not part of connection_kwargs
    # (which is also used to build the producer); inject it here when set.
    if self.client_rack is not None:
        consumer_options["client_rack"] = self.client_rack
    self.builder = partial(aiokafka.AIOKafkaConsumer, **consumer_options)

disconnect async #

disconnect() -> None
Source code in faststream/kafka/configs/broker.py 
75
76
77
78
79
80
81
async def disconnect(self) -> "None":
    if self._admin_client is not None:
        await self._admin_client.close()
        self._admin_client = None

    if not self.consumer_only:
        await self.producer.disconnect()