Skip to content

StreamSub

faststream.redis.schemas.stream_sub.StreamSub #

StreamSub(
    stream: str,
    polling_interval: int | None = None,
    group: str | None = None,
    consumer: str | None = None,
    batch: bool = False,
    no_ack: bool = False,
    last_id: str | None = None,
    maxlen: int | None = None,
    max_records: int | None = None,
    min_idle_time: int | None = None,
    declare: bool = True,
    claim_min_idle_time: int | None = None,
)

Bases: NameRequired

A class to represent a Redis Stream subscriber.

PARAMETER DESCRIPTION
batch

Whether to consume messages in batches or not.

TYPE: bool DEFAULT: False

max_records

Number of messages to consume as one batch.

TYPE: int | None DEFAULT: None

consumer

The consumer unique name

https://redis.io/docs/latest/develop/tools/insight/tutorials/insight-stream-consumer/#run-the-consumer

TYPE: str | None DEFAULT: None

group

The name of consumer group

TYPE: str | None DEFAULT: None

last_id

An Entry ID, which uses to pick up from where it left off after it is restarted.

TYPE: str | None DEFAULT: None

maxlen

Redis Stream maxlen publish option. Remove eldest message if maxlen exceeded.

https://redis.io/docs/latest/develop/data-types/streams/#capped-streams

TYPE: int | None DEFAULT: None

name

The original Redis Stream name.
no_ack

If True, to enable the XREADGROUP NOACK subcommand.

https://redis.io/docs/latest/commands/xreadgroup/#differences-between-xread-and-xreadgroup

TYPE: bool DEFAULT: False

polling_interval

Polling interval in milliseconds.

TYPE: int | None DEFAULT: None

min_idle_time

Minimum idle time in milliseconds for a message to be eligible for claiming via XAUTOCLAIM. Messages that have been pending (unacknowledged) for at least this duration can be reclaimed by this consumer. Only applicable when using consumer groups.

https://redis.io/docs/latest/commands/xautoclaim/

TYPE: int | None DEFAULT: None

declare

Whether to create the stream when creating a consumer group.

TYPE: bool DEFAULT: True

claim_min_idle_time

Redis 8.4+ XREADGROUP CLAIM option (milliseconds). When set, each read also claims messages of the same group that have been pending for at least this duration, so a single subscriber consumes new messages and recovers abandoned ones in one command. Claimed entries arrive first; per-entry metadata is exposed via the idle_times / delivery_counts keys of the raw message. Requires redis-py 7.1.0+ and Redis server 8.4+ (older servers reject the CLAIM option). Mutually exclusive with min_idle_time and no_ack.

https://redis.io/docs/latest/commands/xreadgroup/#the-claim-option

TYPE: int | None DEFAULT: None

Source code in faststream/redis/schemas/stream_sub.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def __init__(
    self,
    stream: str,
    polling_interval: int | None = None,
    group: str | None = None,
    consumer: str | None = None,
    batch: bool = False,
    no_ack: bool = False,
    last_id: str | None = None,
    maxlen: int | None = None,
    max_records: int | None = None,
    min_idle_time: int | None = None,
    declare: bool = True,
    claim_min_idle_time: int | None = None,
) -> None:
    if (group and not consumer) or (not group and consumer):
        msg = "You should specify `group` and `consumer` both"
        raise SetupError(msg)

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

    if last_id is None:
        last_id = ">" if group and consumer else "$"

    if group and consumer and last_id != ">":
        if polling_interval:
            warnings.warn(
                message="`polling_interval` is not supported by consumer group with last_id other than `>`",
                category=RuntimeWarning,
                stacklevel=1,
            )

        if no_ack:
            warnings.warn(
                message="`no_ack` is not supported by consumer group with last_id other than `>`",
                category=RuntimeWarning,
                stacklevel=1,
            )

    if claim_min_idle_time is not None:
        if not REDIS_V710:
            msg = (
                "`claim_min_idle_time` requires redis-py 7.1.0 or newer "
                f"(installed: {_REDIS_VERSION})"
            )
            raise SetupError(msg)

        if min_idle_time is not None:
            msg = (
                "`claim_min_idle_time` (XREADGROUP CLAIM) and `min_idle_time` "
                "(XAUTOCLAIM) are mutually exclusive"
            )
            raise SetupError(msg)

        if no_ack:
            msg = (
                "`claim_min_idle_time` with `no_ack` causes infinite "
                "redelivery: claimed entries stay in the PEL and are "
                "never acknowledged"
            )
            raise SetupError(msg)

        if not group:
            msg = "`claim_min_idle_time` requires `group` and `consumer`"
            raise SetupError(msg)

        if last_id != ">":
            msg = (
                "`claim_min_idle_time` requires last_id `>`: Redis ignores "
                "the CLAIM option for any other id"
            )
            raise SetupError(msg)

    super().__init__(stream)

    self.group = group
    self.consumer = consumer
    self.declare = declare
    self.polling_interval = polling_interval or 100
    self.batch = batch
    self.no_ack = no_ack
    self.last_id = last_id
    self.maxlen = maxlen
    self.max_records = max_records
    self.min_idle_time = min_idle_time
    self.claim_min_idle_time = claim_min_idle_time

group instance-attribute #

group = group

consumer instance-attribute #

consumer = consumer

declare instance-attribute #

declare = declare

polling_interval instance-attribute #

polling_interval = polling_interval or 100

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

min_idle_time instance-attribute #

min_idle_time = min_idle_time

claim_min_idle_time instance-attribute #

claim_min_idle_time = claim_min_idle_time

name instance-attribute #

name = name

add_prefix #

add_prefix(prefix: str) -> StreamSub
Source code in faststream/redis/schemas/stream_sub.py 
165
166
167
168
def add_prefix(self, prefix: str) -> "StreamSub":
    new_stream = deepcopy(self)
    new_stream.name = f"{prefix}{new_stream.name}"
    return new_stream

validate classmethod #

validate(value: str | Self, **kwargs: Any) -> Self

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

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

Factory to create object.

Source code in faststream/_internal/proto.py 
30
31
32
33
34
35
@classmethod
def validate(cls, value: str | Self | None, **kwargs: Any) -> Self | None:
    """Factory to create object."""
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value