Skip to content

ListSub

faststream.redis.ListSub #

ListSub(
    channel: str,
    batch: bool = False,
    max_records: PositiveInt = 10,
    polling_interval: PositiveFloat = 0.1,
)

Bases: NameRequired

Source code in faststream/redis/schemas.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    channel: str,
    batch: bool = False,
    max_records: PositiveInt = 10,
    polling_interval: PositiveFloat = 0.1,
) -> None:
    super().__init__(
        name=channel,
        batch=batch,
        max_records=max_records,
        polling_interval=polling_interval,
    )

batch class-attribute instance-attribute #

batch: bool = False

max_records class-attribute instance-attribute #

max_records: PositiveInt = 10

name class-attribute instance-attribute #

name: str = Field(...)

polling_interval class-attribute instance-attribute #

polling_interval: PositiveFloat = 0.1

records property #

records: Optional[PositiveInt]

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]

RETURNS DESCRIPTION
Optional[NameRequiredCls]

The validated value.
Source code in faststream/broker/schemas.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@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.

    Returns:
        The validated value.

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