Skip to content

PubSub

faststream.redis.schemas.PubSub #

PubSub(channel, pattern=False, polling_interval=1.0)

Bases: NameRequired

A class to represent a Redis PubSub channel.

Source code in faststream/redis/schemas/pub_sub.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(
    self,
    channel: str,
    pattern: bool = False,
    polling_interval: float = 1.0,
) -> None:
    reg, path = compile_path(
        channel,
        replace_symbol="*",
        patch_regex=lambda x: x.replace(r"\*", ".*"),
    )

    if reg is not None:
        pattern = True

    super().__init__(path)

    self.path_regex = reg
    self.pattern = channel if pattern else None
    self.polling_interval = polling_interval

name instance-attribute #

name = name

path_regex instance-attribute #

path_regex = reg

pattern instance-attribute #

pattern = channel if pattern else None

polling_interval instance-attribute #

polling_interval = polling_interval

validate classmethod #

validate(value: Union[str, NameRequiredCls], **kwargs: Any) -> NameRequiredCls

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

validate(value, **kwargs)

Factory to create object.

Source code in faststream/broker/schemas.py 
38
39
40
41
42
43
44
45
46
47
@classmethod
def validate(
    cls: Type[NameRequiredCls],
    value: Union[str, NameRequiredCls, None],
    **kwargs: Any,
) -> Optional[NameRequiredCls]:
    """Factory to create object."""
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value