Skip to content

KafkaRouter

faststream.kafka.shared.router.KafkaRouter #

KafkaRouter(
    prefix: str = "",
    handlers: Sequence[
        BrokerRoute[ConsumerRecord, SendableMessage]
    ] = (),
    **kwargs: Any
)

Bases: BrokerRouter[str, ConsumerRecord]

A class to represent a Kafka router.

METHOD DESCRIPTION
subscriber

decorator for subscribing to topics and handling messages

Initialize the class.

PARAMETER DESCRIPTION
prefix

Prefix string.

TYPE: str DEFAULT: ''

handlers

Sequence of KafkaRoute objects.

TYPE: Sequence[BrokerRoute[ConsumerRecord, SendableMessage]] DEFAULT: ()

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

Source code in faststream/kafka/shared/router.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    prefix: str = "",
    handlers: Sequence[KafkaRoute[ConsumerRecord, SendableMessage]] = (),
    **kwargs: Any,
) -> None:
    """Initialize the class.

    Args:
        prefix (str): Prefix string.
        handlers (Sequence[KafkaRoute[ConsumerRecord, SendableMessage]]): Sequence of KafkaRoute objects.
        **kwargs (Any): Additional keyword arguments.

    """
    for h in handlers:
        h.args = tuple(prefix + x for x in 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(
    *topics: str, **broker_kwargs: Any
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[
        ConsumerRecord, P_HandlerParams, T_HandlerReturn
    ],
]

A function to subscribe to topics.

PARAMETER DESCRIPTION
*topics

variable number of topic names

DEFAULT: ()

**broker_kwargs

keyword arguments for the broker

DEFAULT: {}

RETURNS DESCRIPTION
Callable[[Callable[P_HandlerParams, T_HandlerReturn]], HandlerCallWrapper[ConsumerRecord, P_HandlerParams, T_HandlerReturn]]

A callable function that wraps the handler function
Source code in faststream/kafka/shared/router.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def subscriber(
    self,
    *topics: str,
    **broker_kwargs: Any,
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[ConsumerRecord, P_HandlerParams, T_HandlerReturn],
]:
    """A function to subscribe to topics.

    Args:
        *topics : variable number of topic names
        **broker_kwargs : keyword arguments for the broker

    Returns:
        A callable function that wraps the handler function

    """
    return self._wrap_subscriber(
        *(self.prefix + x for x in topics),
        **broker_kwargs,
    )