Skip to content

NatsBrokerConfig

faststream.nats.configs.NatsBrokerConfig dataclass #

NatsBrokerConfig(
    *,
    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: NatsFastProducer = FakeNatsFastProducer(),
    logger: LoggerState = LoggerState(),
    fd_config: FastDependsConfig = FastDependsConfig(),
    id_generator: IdGenerator = gen_cor_id,
    underlying_driver_annotations: UnderlyingDriverAnnotations = dict(),
    default_driver_annotations: UnderlyingDriverAnnotations = _context_annotations_factory(),
    broker_dependencies: Sequence[Dependant] = (),
    graceful_timeout: float | None = 15.0,
    ack_policy: AckPolicy = (lambda: EMPTY)(),
    extra_context: dict[str, Any] = dict(),
    js_options: JsInitOptions | dict[str, Any] = dict(),
    js_producer: NatsFastProducer = FakeNatsFastProducer(),
    connection_state: BrokerState = BrokerState(),
    kv_declarer: KVBucketDeclarer = KVBucketDeclarer(),
    os_declarer: OSBucketDeclarer = OSBucketDeclarer(),
)

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

underlying_driver_annotations class-attribute instance-attribute #

underlying_driver_annotations: UnderlyingDriverAnnotations = field(
    default_factory=dict
)

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)

resolved_underlying_driver_annotations property #

resolved_underlying_driver_annotations: (
    UnderlyingDriverAnnotations
)

context property #

context: ContextRepo

js_options class-attribute instance-attribute #

js_options: JsInitOptions | dict[str, Any] = field(
    default_factory=dict
)

producer class-attribute instance-attribute #

producer: NatsFastProducer = field(
    default_factory=FakeNatsFastProducer
)

js_producer class-attribute instance-attribute #

js_producer: NatsFastProducer = field(
    default_factory=FakeNatsFastProducer
)

connection_state class-attribute instance-attribute #

connection_state: BrokerState = field(
    default_factory=BrokerState
)

kv_declarer class-attribute instance-attribute #

kv_declarer: KVBucketDeclarer = field(
    default_factory=KVBucketDeclarer
)

os_declarer class-attribute instance-attribute #

os_declarer: OSBucketDeclarer = field(
    default_factory=OSBucketDeclarer
)

default_driver_annotations class-attribute instance-attribute #

default_driver_annotations: UnderlyingDriverAnnotations = (
    field(default_factory=_context_annotations_factory)
)

add_middleware #

add_middleware(middleware: BrokerMiddleware[Any]) -> None
Source code in faststream/_internal/configs/broker.py 
95
96
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 
98
99
def insert_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self.broker_middlewares = (middleware, *self.broker_middlewares)

connect #

connect(connection: Client) -> None
Source code in faststream/nats/configs/broker.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def connect(self, connection: "Client") -> None:
    stream = connection.jetstream(**self.js_options)

    self.producer.connect(
        connection,
        serializer=self.fd_config._serializer,
        codec=self.broker_codec or DefaultCodec(),
    )

    self.js_producer.connect(
        stream,
        serializer=self.fd_config._serializer,
        codec=self.broker_codec or DefaultCodec(),
    )
    self.kv_declarer.connect(stream)
    self.os_declarer.connect(stream)

    self.connection_state.connect(connection, stream)

disconnect #

disconnect() -> None
Source code in faststream/nats/configs/broker.py 
115
116
117
118
119
120
121
def disconnect(self) -> None:
    self.producer.disconnect()
    self.js_producer.disconnect()
    self.kv_declarer.disconnect()
    self.os_declarer.disconnect()

    self.connection_state.disconnect()