Skip to content

PubSub

faststream.redis.schemas.PubSub #

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

Bases: NameRequired

A class to represent a Redis PubSub channel.

Redis PubSub channel parameters.

PARAMETER DESCRIPTION
channel

(str): Redis PubSub channel name.

TYPE: str

pattern

(bool): use pattern matching.

TYPE: bool DEFAULT: False

polling_interval

(float): wait message block.

TYPE: PositiveFloat DEFAULT: 1.0

Source code in faststream/redis/schemas.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    channel: str,
    pattern: bool = False,
    polling_interval: PositiveFloat = 1.0,
) -> None:
    """Redis PubSub channel parameters.

    Args:
        channel: (str): Redis PubSub channel name.
        pattern: (bool): use pattern matching.
        polling_interval: (float): wait message block.
    """
    reg, path = compile_path(
        channel,
        replace_symbol="*",
        patch_regex=lambda x: x.replace(r"\*", ".*"),
    )

    if reg is not None:
        pattern = True

    super().__init__(
        name=path,
        path_regex=reg,
        pattern=pattern,
        polling_interval=polling_interval,
    )

last_id class-attribute instance-attribute #

last_id: str = '$'

model_config class-attribute instance-attribute #

model_config = {'arbitrary_types_allowed': True}

name class-attribute instance-attribute #

name: str = Field(...)

path_regex class-attribute instance-attribute #

path_regex: Optional[Pattern[str]] = None

pattern class-attribute instance-attribute #

pattern: bool = False

polling_interval class-attribute instance-attribute #

polling_interval: PositiveFloat = 1.0

Config #

arbitrary_types_allowed class-attribute instance-attribute #

arbitrary_types_allowed = True

validate classmethod #

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

Validates a value.

PARAMETER DESCRIPTION
value

The value to be validated.

TYPE: Union[str, NameRequiredCls, None]

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
Optional[NameRequiredCls]

The validated value.
Source code in faststream/broker/schemas.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def validate(
    cls: Type[NameRequiredCls],
    value: Union[str, NameRequiredCls, None],
    **kwargs: Any,
) -> Optional[NameRequiredCls]:
    """Validates a value.

    Args:
        value: The value to be validated.
        **kwargs: Additional keyword arguments.

    Returns:
        The validated value.

    """
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value