RedisRouter

faststream.redis.broker.router.RedisRouter #

RedisRouter(
    prefix: str = "",
    handlers: Iterable[RedisRoute] = (),
    *,
    dependencies: Sequence[Dependant] = (),
    middlewares: Sequence[BrokerMiddleware[Any, Any]] = (),
    routers: Iterable[RedisRegistrator] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    include_in_schema: bool | None = None,
    ack_policy: AckPolicy = EMPTY,
)

Bases: RedisRegistrator, BrokerRouter[UnifyRedisDict, RedisRouterConfig]

Includable to RedisBroker router.

Initialize the RedisRouter.

PARAMETER	DESCRIPTION
`prefix`	String prefix to add to all subscribers queues. TYPE: `str` DEFAULT: `''`
`handlers`	Route object to include. TYPE: `Iterable[RedisRoute]` DEFAULT: `()`
`dependencies`	Dependencies list (`[Dependant(),]`) to apply to all routers' publishers/subscribers. TYPE: `Sequence[Dependant]` DEFAULT: `()`
`middlewares`	Router middlewares to apply to all routers' publishers/subscribers. TYPE: `Sequence[BrokerMiddleware[Any, Any]]` DEFAULT: `()`
`routers`	Routers to apply to broker. TYPE: `Iterable[RedisRegistrator]` DEFAULT: `()`
`parser`	Parser to map original IncomingMessage Msg to FastStream one. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`decoder`	Function to decode FastStream msg bytes body to python objects. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`include_in_schema`	Whetever to include operation in AsyncAPI schema or not. TYPE: `bool \| None` DEFAULT: `None`
`ack_policy`	Default acknowledgement policy for all subscribers in this router. Can be overridden at the subscriber level. TYPE: `AckPolicy` DEFAULT: `EMPTY`

Source code in faststream/redis/broker/router.py

def __init__(
    self,
    prefix: str = "",
    handlers: Iterable[RedisRoute] = (),
    *,
    dependencies: Sequence["Dependant"] = (),
    middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (),
    routers: Iterable[RedisRegistrator] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    include_in_schema: bool | None = None,
    ack_policy: "AckPolicy" = EMPTY,
) -> None:
    """Initialize the RedisRouter.

    Args:
        prefix:
            String prefix to add to all subscribers queues.
        handlers:
            Route object to include.
        dependencies:
            Dependencies list (`[Dependant(),]`) to apply to all routers' publishers/subscribers.
        middlewares:
            Router middlewares to apply to all routers' publishers/subscribers.
        routers:
            Routers to apply to broker.
        parser:
            Parser to map original **IncomingMessage** Msg to FastStream one.
        decoder:
            Function to decode FastStream msg bytes body to python objects.
        include_in_schema:
            Whetever to include operation in AsyncAPI schema or not.
        ack_policy:
            Default acknowledgement policy for all subscribers in this router.
            Can be overridden at the subscriber level.
    """
    super().__init__(
        handlers=handlers,
        config=RedisRouterConfig(
            prefix=prefix,
            ack_policy=ack_policy,
            broker_dependencies=dependencies,
            broker_middlewares=middlewares,
            broker_parser=parser,
            broker_decoder=decoder,
            include_in_schema=include_in_schema,
        ),
        routers=routers,
    )

config `instance-attribute` #

config: ConfigComposition[BrokerConfigType] = (
    ConfigComposition(config)
)

routers `instance-attribute` #

routers: list[Registrator[MsgType, Any]] = []

subscribers `property` #

subscribers: list[SubscriberUsecase[MsgType]]

publishers `property` #

publishers: list[PublisherUsecase]

parent `property` `writable` #

parent: Registrator[MsgType, Any] | None

add_middleware #

add_middleware(
    middleware: BrokerMiddleware[Any, Any],
) -> None

Append BrokerMiddleware to the end of middlewares list.

Current middleware will be used as a most inner of the stack.

Source code in faststream/_internal/broker/registrator.py

def add_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None:
    """Append BrokerMiddleware to the end of middlewares list.

    Current middleware will be used as a most inner of the stack.
    """
    self.config.add_middleware(middleware)

insert_middleware #

insert_middleware(
    middleware: BrokerMiddleware[Any, Any],
) -> None

Insert BrokerMiddleware to the start of middlewares list.

Current middleware will be used as a most outer of the stack.

Source code in faststream/_internal/broker/registrator.py

def insert_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None:
    """Insert BrokerMiddleware to the start of middlewares list.

    Current middleware will be used as a most outer of the stack.
    """
    self.config.insert_middleware(middleware)

subscriber #

subscriber(
    channel: Union[PubSub, str] = ...,
    *,
    list: None = None,
    stream: None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> ChannelSubscriber

subscriber(
    channel: Union[PubSub, str] = ...,
    *,
    list: None = None,
    stream: None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: int = ...,
) -> ChannelConcurrentSubscriber

subscriber(
    channel: None = None,
    *,
    list: Union[str, ListSub[Literal[False]]] = ...,
    stream: None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> ListSubscriber

subscriber(
    channel: None = None,
    *,
    list: ListSub[Literal[True]] = ...,
    stream: None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> ListBatchSubscriber

subscriber(
    channel: None = None,
    *,
    list: Union[ListSub, str] = ...,
    stream: None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> Union[ListSubscriber, ListBatchSubscriber]

subscriber(
    channel: None = None,
    *,
    list: Union[ListSub, str] = ...,
    stream: None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: int = ...,
) -> ListConcurrentSubscriber

subscriber(
    channel: None = None,
    *,
    list: None = None,
    stream: Union[str, StreamSub[Literal[False]]] = ...,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> StreamSubscriber

subscriber(
    channel: None = None,
    *,
    list: None = None,
    stream: StreamSub[Literal[True]] = ...,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> StreamBatchSubscriber

subscriber(
    channel: None = None,
    *,
    list: None = None,
    stream: Union[StreamSub, str] = ...,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: None = None,
) -> Union[StreamSubscriber, StreamBatchSubscriber]

subscriber(
    channel: None = None,
    *,
    list: None = None,
    stream: Union[StreamSub, str] = ...,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: int = ...,
) -> StreamConcurrentSubscriber

subscriber(
    channel: Union[PubSub, str, None] = None,
    *,
    list: Union[ListSub, str, None] = None,
    stream: Union[StreamSub, str, None] = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: int | None = None,
) -> LogicSubscriber

subscriber(
    channel: Union[PubSub, str, None] = None,
    *,
    list: Union[ListSub, str, None] = None,
    stream: Union[StreamSub, str, None] = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    codec: Optional[CodecProto] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: int | None = None,
) -> LogicSubscriber

Subscribe a handler to a RabbitMQ queue.

PARAMETER	DESCRIPTION
`channel`	Redis PubSub object name to send message. TYPE: `Union[PubSub, str, None]` DEFAULT: `None`
`list`	Redis List object name to send message. TYPE: `Union[ListSub, str, None]` DEFAULT: `None`
`stream`	Redis Stream object name to send message. TYPE: `Union[StreamSub, str, None]` DEFAULT: `None`
`ack_policy`	Acknowledgement policy for message processing. TYPE: `AckPolicy` DEFAULT: `EMPTY`
`dependencies`	Dependencies list (`[Depends(),]`) to apply to the subscriber. TYPE: `Sequence[Dependant]` DEFAULT: `()`
`parser`	Parser to map original IncomingMessage Msg to FastStream one. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`decoder`	Function to decode FastStream msg bytes body to python objects. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`codec`	Custom codec object. TYPE: `Optional[CodecProto]` DEFAULT: `None`
`no_reply`	Whether to disable FastStream RPC and Reply To auto responses or not. TYPE: `bool` DEFAULT: `False`
`message_format`	Which format to use when parsing messages. TYPE: `type[MessageFormat] \| None` DEFAULT: `None`
`persistent`	Whether to make the subscriber persistent or not. TYPE: `bool` DEFAULT: `True`
`max_workers`	Number of workers to process messages concurrently. TYPE: `int \| None` DEFAULT: `None`
`title`	AsyncAPI subscriber object title. TYPE: `str \| None` DEFAULT: `None`
`description`	AsyncAPI subscriber object description. Uses decorated docstring as default. TYPE: `str \| None` DEFAULT: `None`
`include_in_schema`	Whether to include operation in AsyncAPI schema or not. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`SubscriberType`	The subscriber object. TYPE: `LogicSubscriber`

Source code in faststream/redis/broker/registrator.py

@override
def subscriber(
    self,
    channel: Union["PubSub", str, None] = None,
    *,
    list: Union["ListSub", str, None] = None,
    stream: Union["StreamSub", str, None] = None,
    # broker arguments
    dependencies: Sequence["Dependant"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    codec: Optional["CodecProto"] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    message_format: type["MessageFormat"] | None = None,
    persistent: bool = True,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    max_workers: int | None = None,
) -> "LogicSubscriber":
    """Subscribe a handler to a RabbitMQ queue.

    Args:
        channel: Redis PubSub object name to send message.
        list: Redis List object name to send message.
        stream: Redis Stream object name to send message.
        ack_policy: Acknowledgement policy for message processing.
        dependencies: Dependencies list (`[Depends(),]`) to apply to the subscriber.
        parser: Parser to map original **IncomingMessage** Msg to FastStream one.
        decoder: Function to decode FastStream msg bytes body to python objects.
        codec: Custom codec object.
        no_reply: Whether to disable **FastStream** RPC and Reply To auto responses or not.
        message_format: Which format to use when parsing messages.
        persistent: Whether to make the subscriber persistent or not.
        max_workers: Number of workers to process messages concurrently.
        title: AsyncAPI subscriber object title.
        description: AsyncAPI subscriber object description. Uses decorated docstring as default.
        include_in_schema: Whether to include operation in AsyncAPI schema or not.

    Returns:
        SubscriberType: The subscriber object.
    """
    subscriber = create_subscriber(
        channel=channel,
        list=list,
        stream=stream,
        # subscriber args
        max_workers=max_workers or 1,
        no_reply=no_reply,
        ack_policy=ack_policy,
        message_format=message_format,
        config=cast("RedisBrokerConfig", self.config),
        # AsyncAPI
        title_=title,
        description_=description,
        include_in_schema=include_in_schema,
    )

    super().subscriber(subscriber, persistent=persistent)

    return subscriber.add_call(
        parser_=parser or self._parser,
        decoder_=decoder or self._decoder,
        codec_=codec,
        dependencies_=dependencies,
    )

publisher #

publisher(
    channel: None = None,
    *,
    list: None = None,
    stream: Union[StreamSub, str] = ...,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> StreamPublisher

publisher(
    channel: None = None,
    *,
    list: Union[str, ListSub[Literal[False]]] = ...,
    stream: None = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> ListPublisher

publisher(
    channel: None = None,
    *,
    list: ListSub[Literal[True]] = ...,
    stream: None = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> ListBatchPublisher

publisher(
    channel: None = None,
    *,
    list: Union[ListSub, str] = ...,
    stream: None = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> Union[ListPublisher, ListBatchPublisher]

publisher(
    channel: Union[PubSub, str] = ...,
    *,
    list: None = None,
    stream: None = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> ChannelPublisher

publisher(
    channel: Union[PubSub, str, None] = None,
    *,
    list: Union[ListSub, str, None] = None,
    stream: Union[StreamSub, str, None] = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> LogicPublisher

publisher(
    channel: Union[PubSub, str, None] = None,
    *,
    list: Union[ListSub, str, None] = None,
    stream: Union[StreamSub, str, None] = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type[MessageFormat] | None = None,
    persistent: bool = True,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> LogicPublisher

Creates long-living and AsyncAPI-documented publisher object.

You can use it as a handler decorator (handler should be decorated by @broker.subscriber(...) too) - @broker.publisher(...). In such case publisher will publish your handler return value.

Or you can create a publisher object to call it lately - broker.publisher(...).publish(...).

PARAMETER	DESCRIPTION
`channel`	Redis PubSub object name to send message. TYPE: `Union[PubSub, str, None]` DEFAULT: `None`
`list`	Redis List object name to send message. TYPE: `Union[ListSub, str, None]` DEFAULT: `None`
`stream`	Redis Stream object name to send message. TYPE: `Union[StreamSub, str, None]` DEFAULT: `None`
`headers`	Message headers to store meta-information. Can be overridden by `publish.headers` if specified. TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`reply_to`	Reply message destination PubSub object name. TYPE: `str` DEFAULT: `''`
`message_format`	Which format to use when parsing messages. TYPE: `type[MessageFormat] \| None` DEFAULT: `None`
`title`	AsyncAPI publisher object title. TYPE: `str \| None` DEFAULT: `None`
`description`	AsyncAPI publisher object description. TYPE: `str \| None` DEFAULT: `None`
`schema`	AsyncAPI publishing message type. Should be any python-native object annotation or `pydantic.BaseModel`. TYPE: `Any \| None` DEFAULT: `None`
`include_in_schema`	Whether to include operation in AsyncAPI schema or not. TYPE: `bool` DEFAULT: `True`
`persistent`	Whether to make the publisher persistent or not. TYPE: `bool` DEFAULT: `True`

Source code in faststream/redis/broker/registrator.py

@override
def publisher(
    self,
    channel: Union["PubSub", str, None] = None,
    *,
    list: Union["ListSub", str, None] = None,
    stream: Union["StreamSub", str, None] = None,
    headers: dict[str, Any] | None = None,
    reply_to: str = "",
    message_format: type["MessageFormat"] | None = None,
    persistent: bool = True,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> "LogicPublisher":
    """Creates long-living and AsyncAPI-documented publisher object.

    You can use it as a handler decorator (handler should be decorated by `@broker.subscriber(...)` too) - `@broker.publisher(...)`.
    In such case publisher will publish your handler return value.

    Or you can create a publisher object to call it lately - `broker.publisher(...).publish(...)`.

    Args:
        channel: Redis PubSub object name to send message.
        list: Redis List object name to send message.
        stream: Redis Stream object name to send message.
        headers: Message headers to store meta-information. Can be overridden
            by `publish.headers` if specified.
        reply_to: Reply message destination PubSub object name.
        message_format: Which format to use when parsing messages.
        title: AsyncAPI publisher object title.
        description: AsyncAPI publisher object description.
        schema: AsyncAPI publishing message type. Should be any python-native
            object annotation or `pydantic.BaseModel`.
        include_in_schema: Whether to include operation in AsyncAPI schema or not.
        persistent: Whether to make the publisher persistent or not.
    """
    publisher = create_publisher(
        channel=channel,
        list=list,
        stream=stream,
        headers=headers,
        reply_to=reply_to,
        # Specific
        config=cast("RedisBrokerConfig", self.config),
        message_format=message_format,
        # AsyncAPI
        title_=title,
        description_=description,
        schema_=schema,
        include_in_schema=include_in_schema,
    )
    super().publisher(publisher, persistent=persistent)
    return publisher

include_router #

include_router(
    router: RedisRegistrator,
    *,
    prefix: str = "",
    dependencies: Sequence[Dependant] = (),
    middlewares: Sequence[BrokerMiddleware[Any, Any]] = (),
    include_in_schema: bool | None = None,
) -> None

Source code in faststream/redis/broker/registrator.py

@override
def include_router(
    self,
    router: "RedisRegistrator",  # type: ignore[override]
    *,
    prefix: str = "",
    dependencies: Sequence["Dependant"] = (),
    middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (),
    include_in_schema: bool | None = None,
) -> None:
    if not isinstance(router, RedisRegistrator):
        msg = (
            f"Router must be an instance of RedisRegistrator, "
            f"got {type(router).__name__} instead"
        )
        raise SetupError(msg)

    super().include_router(
        router,
        prefix=prefix,
        dependencies=dependencies,
        middlewares=middlewares,
        include_in_schema=include_in_schema,
    )

include_routers #

include_routers(
    *routers: Registrator[MsgType, Any],
) -> None

Includes routers in the object.

Source code in faststream/_internal/broker/registrator.py

def include_routers(
    self,
    *routers: "Registrator[MsgType, Any]",
) -> None:
    """Includes routers in the object."""
    for r in routers:
        self.include_router(r)

RedisRouter

faststream.redis.broker.router.RedisRouter #

config instance-attribute #

routers instance-attribute #

subscribers property #

publishers property #

parent property writable #

add_middleware #

insert_middleware #

subscriber #

publisher #

include_router #

include_routers #

config `instance-attribute` #

routers `instance-attribute` #

subscribers `property` #

publishers `property` #

parent `property` `writable` #