Skip to content

NatsRouter

faststream.nats.shared.router.NatsRouter #

NatsRouter(
    prefix: str = "",
    handlers: Sequence[NatsRoute] = (),
    **kwargs: Any
)

Bases: BrokerRouter[str, Msg]

A class to represent a NATS router.

Initialize the NATS router.

PARAMETER DESCRIPTION
prefix

The prefix.

TYPE: str DEFAULT: ''

handlers

The handlers.

TYPE: Sequence[NatsRoute] DEFAULT: ()

**kwargs

The keyword arguments.

TYPE: Any DEFAULT: {}

Source code in faststream/nats/shared/router.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    prefix: str = "",
    handlers: Sequence[NatsRoute[Msg, SendableMessage]] = (),
    **kwargs: Any,
) -> None:
    """Initialize the NATS router.

    Args:
        prefix: The prefix.
        handlers: The handlers.
        **kwargs: The keyword arguments.
    """
    for h in handlers:
        if not (subj := h.kwargs.pop("subject", None)):
            subj, h.args = h.args[0], h.args[1:]
        h.args = (prefix + subj, *h.args)
    super().__init__(prefix, handlers, **kwargs)

include_in_schema instance-attribute #

include_in_schema = include_in_schema

prefix instance-attribute #

prefix: str = prefix

include_router #

include_router(
    router: BrokerRouter[PublisherKeyType, MsgType]
) -> None

Includes a router in the current object.

PARAMETER DESCRIPTION
router

The router to be included.

TYPE: BrokerRouter[PublisherKeyType, MsgType]

RETURNS DESCRIPTION
None

None
Source code in faststream/broker/router.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def include_router(self, router: "BrokerRouter[PublisherKeyType, MsgType]") -> None:
    """Includes a router in the current object.

    Args:
        router: The router to be included.

    Returns:
        None

    """
    for h in router._handlers:
        self.subscriber(*h.args, **h.kwargs)(h.call)

    for p in router._publishers.values():
        p = self._update_publisher_prefix(self.prefix, p)
        key = self._get_publisher_key(p)
        self._publishers[key] = self._publishers.get(key, p)

include_routers #

include_routers(
    *routers: BrokerRouter[PublisherKeyType, MsgType]
) -> None

Includes routers in the object.

PARAMETER DESCRIPTION
*routers

Variable length argument list of routers to include.

TYPE: BrokerRouter[PublisherKeyType, MsgType] DEFAULT: ()

RETURNS DESCRIPTION
None

None
Source code in faststream/broker/router.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def include_routers(
    self, *routers: "BrokerRouter[PublisherKeyType, MsgType]"
) -> None:
    """Includes routers in the object.

    Args:
        *routers: Variable length argument list of routers to include.

    Returns:
        None

    """
    for r in routers:
        self.include_router(r)

publisher abstractmethod #

publisher(
    subj: str, *args: Any, **kwargs: Any
) -> BasePublisher[MsgType]

Publishes a message.

PARAMETER DESCRIPTION
subj

Subject of the message

TYPE: str

*args

Additional arguments

TYPE: Any DEFAULT: ()

**kwargs

Additional keyword arguments

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
BasePublisher[MsgType]

The published message
RAISES DESCRIPTION
NotImplementedError

If the method is not implemented
Source code in faststream/broker/router.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
@abstractmethod
def publisher(
    self,
    subj: str,
    *args: Any,
    **kwargs: Any,
) -> BasePublisher[MsgType]:
    """Publishes a message.

    Args:
        subj: Subject of the message
        *args: Additional arguments
        **kwargs: Additional keyword arguments

    Returns:
        The published message

    Raises:
        NotImplementedError: If the method is not implemented

    """
    raise NotImplementedError()

subscriber #

subscriber(
    subject: str, **broker_kwargs: Any
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[
        Msg, P_HandlerParams, T_HandlerReturn
    ],
]
Source code in faststream/nats/shared/router.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
@override
def subscriber(  # type: ignore[override]
    self,
    subject: str,
    **broker_kwargs: Any,
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[Msg, P_HandlerParams, T_HandlerReturn],
]:
    return self._wrap_subscriber(
        self.prefix + subject,
        **broker_kwargs,
    )