KafkaBrokerConfig

faststream.confluent.configs.KafkaBrokerConfig dataclass

KafkaBrokerConfig(
    *,
    prefix="",
    include_in_schema=True,
    broker_middlewares=(),
    broker_parser=None,
    broker_decoder=None,
    producer=FakeConfluentFastProducer(),
    logger=LoggerState(),
    fd_config=FastDependsConfig(),
    broker_dependencies=(),
    graceful_timeout=None,
    extra_context=dict(),
    connection_config=ConfluentFastConfig(),
    admin=AdminService(),
    client_id=SERVICE_NAME,
)

Bases: BrokerConfig

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)

connection_config class-attribute instance-attribute

connection_config = field(
    default_factory=ConfluentFastConfig
)

admin class-attribute instance-attribute

admin = field(default_factory=AdminService)

client_id class-attribute instance-attribute

client_id = SERVICE_NAME

builder class-attribute instance-attribute

builder = field(init=False)

producer class-attribute instance-attribute

producer = field(default_factory=FakeConfluentFastProducer)

add_middleware

add_middleware(middleware)

Source code in faststream/_internal/configs/broker.py

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

def insert_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self.broker_middlewares = (middleware, *self.broker_middlewares)

connect async

connect()

Source code in faststream/confluent/configs/broker.py

async def connect(self) -> "None":
    native_producer = AsyncConfluentProducer(
        config=self.connection_config,
        logger=self.logger,
    )
    self.producer.connect(native_producer, serializer=self.fd_config._serializer)
    await self.admin.connect(self.connection_config)

disconnect async

disconnect()

Source code in faststream/confluent/configs/broker.py

async def disconnect(self) -> "None":
    await self.producer.disconnect()
    await self.admin.disconnect()