Skip to content

KafkaBrokerConfig

faststream.kafka.configs.broker.KafkaBrokerConfig dataclass #

KafkaBrokerConfig(
    *,
    prefix="",
    include_in_schema=True,
    broker_middlewares=(),
    broker_parser=None,
    broker_decoder=None,
    producer=FakeAioKafkaFastProducer(),
    logger=LoggerState(),
    fd_config=FastDependsConfig(),
    broker_dependencies=(),
    graceful_timeout=None,
    extra_context=dict(),
    builder=lambda: None,
    client_id=SERVICE_NAME,
    _admin_client=None,
)

Bases: BrokerConfig

producer class-attribute instance-attribute #

producer = field(default_factory=FakeAioKafkaFastProducer)

builder class-attribute instance-attribute #

builder = lambda: None

client_id class-attribute instance-attribute #

client_id = SERVICE_NAME

admin_client property #

admin_client

prefix class-attribute instance-attribute #

prefix = ''

include_in_schema class-attribute instance-attribute #

include_in_schema = True

broker_middlewares class-attribute instance-attribute #

broker_middlewares = ()

broker_parser class-attribute instance-attribute #

broker_parser = None

broker_decoder class-attribute instance-attribute #

broker_decoder = None

logger class-attribute instance-attribute #

logger = field(default_factory=LoggerState)

fd_config class-attribute instance-attribute #

fd_config = field(default_factory=FastDependsConfig)

broker_dependencies class-attribute instance-attribute #

broker_dependencies = ()

graceful_timeout class-attribute instance-attribute #

graceful_timeout = None

extra_context class-attribute instance-attribute #

extra_context = field(default_factory=dict)

connect async #

connect(**connection_kwargs)
Source code in faststream/kafka/configs/broker.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
async def connect(self, **connection_kwargs: Any) -> "None":
    producer = aiokafka.AIOKafkaProducer(**connection_kwargs)
    await self.producer.connect(producer, serializer=self.fd_config._serializer)

    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,
    )
    self.builder = partial(aiokafka.AIOKafkaConsumer, **consumer_options)

disconnect async #

disconnect()
Source code in faststream/kafka/configs/broker.py 
58
59
60
61
62
63
async def disconnect(self) -> "None":
    if self._admin_client is not None:
        await self._admin_client.close()
        self._admin_client = None

    await self.producer.disconnect()

add_middleware #

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

insert_middleware #

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