Skip to content

RabbitBrokerConfig

faststream.rabbit.configs.broker.RabbitBrokerConfig dataclass #

RabbitBrokerConfig(
    *,
    prefix="",
    include_in_schema=True,
    broker_middlewares=(),
    broker_parser=None,
    broker_decoder=None,
    producer=FakeAioPikaFastProducer(),
    logger=LoggerState(),
    fd_config=FastDependsConfig(),
    broker_dependencies=(),
    graceful_timeout=None,
    extra_context=dict(),
    channel_manager=FakeChannelManager(),
    declarer=FakeRabbitDeclarer(),
    virtual_host="",
    app_id=None,
)

Bases: BrokerConfig

channel_manager class-attribute instance-attribute #

channel_manager = field(default_factory=FakeChannelManager)

declarer class-attribute instance-attribute #

declarer = field(default_factory=FakeRabbitDeclarer)

producer class-attribute instance-attribute #

producer = field(default_factory=FakeAioPikaFastProducer)

virtual_host class-attribute instance-attribute #

virtual_host = ''

app_id class-attribute instance-attribute #

app_id = None

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 #

connect(connection)
Source code in faststream/rabbit/configs/broker.py 
28
29
30
def connect(self, connection: "RobustConnection") -> None:
    self.channel_manager.connect(connection)
    self.producer.connect(serializer=self.fd_config._serializer)

disconnect #

disconnect()
Source code in faststream/rabbit/configs/broker.py 
32
33
34
35
def disconnect(self) -> None:
    self.channel_manager.disconnect()
    self.declarer.disconnect()
    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)