Skip to content

RedisRouter

faststream.redis.shared.router.RedisRouter #

RedisRouter(prefix: str = '', handlers: Sequence[BrokerRoute[AnyRedisDict, SendableMessage]] = (), **kwargs: Any)

Bases: BrokerRouter[int, AnyRedisDict]

A class to represent a Redis router.

Initialize the Redis router.

PARAMETER DESCRIPTION
prefix

The prefix.

TYPE: str DEFAULT: ''

handlers

The handlers.

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

**kwargs

The keyword arguments.

TYPE: Any DEFAULT: {}

Source code in faststream/redis/shared/router.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    prefix: str = "",
    handlers: Sequence[RedisRoute[AnyRedisDict, SendableMessage]] = (),
    **kwargs: Any,
) -> None:
    """Initialize the Redis router.

    Args:
        prefix: The prefix.
        handlers: The handlers.
        **kwargs: The keyword arguments.
    """
    for h in handlers:
        if not (channel := h.kwargs.pop("channel", None)):
            if list := h.kwargs.pop("list", None):
                h.kwargs["list"] = prefix + list
                continue

            elif stream := h.kwargs.pop("stream", None):
                h.kwargs["stream"] = prefix + stream
                continue

            channel, h.args = h.args[0], h.args[1:]

        h.args = (prefix + channel, *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(channel: Union[Channel, PubSub, None] = None, *, list: Union[Channel, ListSub, None] = None, stream: Union[Channel, StreamSub, None] = None, **broker_kwargs: Any) -> Callable[[Callable[P_HandlerParams, T_HandlerReturn]], HandlerCallWrapper[AnyRedisDict, P_HandlerParams, T_HandlerReturn]]
Source code in faststream/redis/shared/router.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@override
def subscriber(  # type: ignore[override]
    self,
    channel: Union[Channel, PubSub, None] = None,
    *,
    list: Union[Channel, ListSub, None] = None,
    stream: Union[Channel, StreamSub, None] = None,
    **broker_kwargs: Any,
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[AnyRedisDict, P_HandlerParams, T_HandlerReturn],
]:
    channel = PubSub.validate(channel)
    list = ListSub.validate(list)
    stream = StreamSub.validate(stream)

    return self._wrap_subscriber(
        channel=model_copy(channel, update={"name": self.prefix + channel.name})
        if channel
        else None,
        list=model_copy(list, update={"name": self.prefix + list.name})
        if list
        else None,
        stream=model_copy(stream, update={"name": self.prefix + stream.name})
        if stream
        else None,
        **broker_kwargs,
    )