Skip to content

StreamSub

faststream.redis.schemas.stream_sub.StreamSub #

StreamSub(
    stream,
    polling_interval=100,
    group=None,
    consumer=None,
    batch=False,
    no_ack=False,
    last_id=None,
    maxlen=None,
    max_records=None,
)

Bases: NameRequired

A class to represent a Redis Stream subscriber.

Source code in faststream/redis/schemas/stream_sub.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    stream: str,
    polling_interval: Optional[int] = 100,
    group: Optional[str] = None,
    consumer: Optional[str] = None,
    batch: bool = False,
    no_ack: bool = False,
    last_id: Optional[str] = None,
    maxlen: Optional[int] = None,
    max_records: Optional[int] = None,
) -> None:
    if (group and not consumer) or (not group and consumer):
        raise SetupError("You should specify `group` and `consumer` both")

    if group and consumer and no_ack:
        warnings.warn(
            message="`no_ack` has no effect with consumer group",
            category=RuntimeWarning,
            stacklevel=1,
        )

    if last_id is None:
        last_id = "$"

    super().__init__(stream)

    self.group = group
    self.consumer = consumer
    self.polling_interval = polling_interval
    self.batch = batch
    self.no_ack = no_ack
    self.last_id = last_id
    self.maxlen = maxlen
    self.max_records = max_records

name instance-attribute #

name = name

group instance-attribute #

group = group

consumer instance-attribute #

consumer = consumer

polling_interval instance-attribute #

polling_interval = polling_interval

batch instance-attribute #

batch = batch

no_ack instance-attribute #

no_ack = no_ack

last_id instance-attribute #

last_id = last_id

maxlen instance-attribute #

maxlen = maxlen

max_records instance-attribute #

max_records = max_records

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