Skip to content

ListSub

faststream.redis.schemas.list_sub.ListSub #

ListSub(
    list_name: str,
    batch: Literal[False] = False,
    max_records: int = 10,
    polling_interval: float = 0.1,
)

ListSub(
    list_name: str,
    batch: Literal[True],
    max_records: int = 10,
    polling_interval: float = 0.1,
)

ListSub(
    list_name: str,
    batch: bool,
    max_records: int = 10,
    polling_interval: float = 0.1,
)

ListSub(
    list_name: str,
    batch: bool = False,
    max_records: int = 10,
    polling_interval: float = 0.1,
)

Bases: NameRequired, Generic[BatchT_co]

A class to represent a Redis List subscriber.

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

    self.batch = batch
    self.max_records = max_records
    self.polling_interval = polling_interval

batch instance-attribute #

batch = batch

max_records instance-attribute #

max_records = max_records

polling_interval instance-attribute #

polling_interval = polling_interval

records cached property #

records: int | None

name instance-attribute #

name = name

validate classmethod #

validate(
    value: str | ListSub[BatchT], **kwargs: Any
) -> ListSub[BatchT]

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

validate(
    value: str | ListSub[Any] | None, **kwargs: Any
) -> ListSub[Any] | None
Source code in faststream/redis/schemas/list_sub.py 
79
80
81
82
83
84
85
86
@classmethod
def validate(
    cls, value: "str | ListSub[Any] | None", **kwargs: Any
) -> "ListSub[Any] | None":
    # `Self` would resolve to the non-batch parametrization of the first `__init__`
    if isinstance(value, str):
        return cls(value, **kwargs)
    return value

add_prefix #

add_prefix(prefix: str) -> Self
Source code in faststream/redis/schemas/list_sub.py 
88
89
90
91
def add_prefix(self, prefix: str) -> Self:
    new_list = deepcopy(self)
    new_list.name = f"{prefix}{new_list.name}"
    return new_list