Skip to content

PubSub

faststream.redis.schemas.PubSub #

PubSub(
    channel: str,
    pattern: bool = False,
    polling_interval: float = 1.0,
)

Bases: NameRequired

A class to represent a Redis PubSub channel.

Source code in faststream/redis/schemas/pub_sub.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    channel: str,
    pattern: bool = False,
    polling_interval: float = 1.0,
) -> None:
    address = Address(channel, REDIS_ADDRESS_SYNTAX)

    if address.regex is not None or "*" in channel:
        pattern = True

    # `name` is `NameRequired`'s writable contract, so it mirrors the Broker
    # address rather than deriving from it. Both build points keep it in step.
    super().__init__(address.broker_address)

    self.address = address
    self.pattern = pattern
    self.polling_interval = polling_interval

address instance-attribute #

address: Address = address

The channel this endpoint was declared with, and its Broker address.

name instance-attribute #

name: str

The channel Redis is (p)subscribed to, e.g. logs.* for logs.{level}.

pattern instance-attribute #

pattern: bool = pattern

Whether to subscribe with psubscribe rather than subscribe.

polling_interval instance-attribute #

polling_interval = polling_interval

path_regex property #

path_regex: Pattern[str] | None

validate classmethod #

validate(value: str | Self, **kwargs: Any) -> Self

validate(value: None, **kwargs: Any) -> None

validate(
    value: str | Self | None, **kwargs: Any
) -> Self | None

Factory to create object.

Source code in faststream/_internal/proto.py 
30
31
32
33
34
35
@classmethod
def validate(cls, value: str | Self | None, **kwargs: Any) -> Self | None:
    """Factory to create object."""
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value

add_prefix #

add_prefix(prefix: str) -> PubSub
Source code in faststream/redis/schemas/pub_sub.py 
55
56
57
58
59
def add_prefix(self, prefix: str) -> "PubSub":
    new_ch = deepcopy(self)
    new_ch.address = new_ch.address.add_prefix(prefix)
    new_ch.name = new_ch.address.broker_address
    return new_ch