Skip to content

KafkaBrokerConfig

faststream.confluent.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,
    producer: AsyncConfluentFastProducer = FakeConfluentFastProducer(),
    logger: LoggerState = LoggerState(),
    fd_config: FastDependsConfig = FastDependsConfig(),
    broker_dependencies: Iterable[Dependant] = (),
    graceful_timeout: float | None = None,
    extra_context: dict[str, Any] = dict(),
    connection_config: ConfluentFastConfig = ConfluentFastConfig(),
    admin: AdminService = AdminService(),
    client_id: str | None = SERVICE_NAME,
)

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

logger class-attribute instance-attribute #

logger: LoggerState = field(default_factory=LoggerState)

fd_config class-attribute instance-attribute #

fd_config: FastDependsConfig = field(
    default_factory=FastDependsConfig
)

broker_dependencies class-attribute instance-attribute #

broker_dependencies: Iterable[Dependant] = ()

graceful_timeout class-attribute instance-attribute #

graceful_timeout: float | None = None

extra_context class-attribute instance-attribute #

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

connection_config class-attribute instance-attribute #

connection_config: ConfluentFastConfig = field(
    default_factory=ConfluentFastConfig
)

admin class-attribute instance-attribute #

admin: AdminService = field(default_factory=AdminService)

client_id class-attribute instance-attribute #

client_id: str | None = SERVICE_NAME

builder class-attribute instance-attribute #

builder: Callable[..., AsyncConfluentConsumer] = field(
    init=False
)

producer class-attribute instance-attribute #

producer: AsyncConfluentFastProducer = field(
    default_factory=FakeConfluentFastProducer
)

add_middleware #

add_middleware(middleware: BrokerMiddleware[Any]) -> None
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: BrokerMiddleware[Any],
) -> None
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)

connect async #

connect() -> None
Source code in faststream/confluent/configs/broker.py 
59
60
61
62
63
64
65
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() -> None
Source code in faststream/confluent/configs/broker.py 
67
68
69
async def disconnect(self) -> "None":
    await self.producer.disconnect()
    await self.admin.disconnect()