🚀 New issue to ag2ai/faststream by @IvanKirpichnikov
📝 Feature: Add `RedisBatchStreamMessage` annotated alias (#3004)


Add to faststream/redis/annotations.py the alias for RedisBatchStreamMessage is similar to the aliases for RedisMessage, RedisChannelMessage, etc.


#enhancement #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Feature: `StreamSub(declare=False)` to opt out of Redis Stream creation (#3013)


Is your feature request related to a problem? Please describe.

FastStream creates the Redis Stream for you whenever a subscriber uses a consumer group — mkstream=True is hardcoded and there is no way to turn it off:

faststream/faststream/redis/subscriber/usecases/stream_subscriber.py

Lines 133 to 138 in 2e36197

That is a fine default, but it means a typo in a stream name silently provisions a new empty stream instead of failing, and a service that is only supposed to consume a stream provisioned elsewhere has no way to say so. Every other broker in FastStream already has this opt-out — RabbitQueue(declare=False), RabbitExchange(declare=False), JStream(declare=False), KvWatch(declare=False), ObjWatch(declare=False). Redis is the only one missing it. The same knob is being added to Kafka in #1827.

Describe the solution you'd like

A declare: bool = True argument on StreamSub, passed through as mkstream=not declare. Default behaviour does not change.

Feature code example

from faststream import FastStream
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)


# unchanged default — stream is created if missing
@broker.subscriber(stream=StreamSub("orders", group="g", consumer="c"))
async def owned(msg: str) -> None: ...


# provisioned elsewhere — fail fast instead of creating an empty stream
@broker.subscriber(
    stream=StreamSub("external-events", group="g", consumer="c", declare=False)
)
async def consumed(msg: str) -> None: ...

Semantics

Redis only offers the "check, do not create" form, so declare=False fails when the stream is absent — the same as RabbitMQ's declare=False (AMQP passive=True), and unlike NATS' variant which just skips creation. Verified against redis:alpine:

mkstream=False -> ResponseError: The XGROUP subcommand requires the key to exist.
                  Note that for CREATE you may want to use the MKSTREAM option to
                  create an empty stream automatically.
exists after mkstream=False: 0
mkstream=True  -> exists after: 1

The existing except ResponseError only swallows "already exists", so this error propagates as-is today — which is the wanted behaviour. Wrapping it in a FastStream error with a message pointing at declare=False would be a nice touch, not a requirement.

Implementation notes

1. Add declare: bool = True to StreamSub.__init__ and to __slots__ —

 faststream/faststream/redis/schemas/stream_sub.py

 Lines 44 to 69 in 2e36197

 — plus an Args: entry in the class docstring.
2. Pass mkstream=not stream.declare at the xgroup_create call linked above.
3. declare only has an effect on the consumer-group path — that xgroup_create runs only when group and consumer are both set. Without a group the subscriber uses XREAD, which never creates anything. StreamSub already emits RuntimeWarning for argument combinations that have no effect (no_ack with a group, polling_interval with last_id != ">"); declare=False without a group deserves the same treatment.
4. Tests go in tests/brokers/redis/ — one real-broker case asserting the stream is not created and the error surfaces, one asserting the default still creates it.

Out of scope: publishing. XADD always creates the stream and Redis exposes no flag to prevent it, so there is no publisher-side equivalent.

Additional context

Prior art to copy the naming and docstring style from: RabbitQueue.declare and JStream.declare.


#enhancement #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Bug: `redis.asyncio.Redis` type hint is not injected and fails validation (#3065)


Describe the bug

Annotating a handler argument with redis.asyncio.Redis — the obvious type for the connection — does not inject it. Instead the argument is treated as a message field and fails validation. The working annotation is faststream.redis.annotations.Redis, which is a different symbol with the same name.

The two names are identical, the import paths differ, and nothing in the failure points at the fix.

How to reproduce

from redis.asyncio import Redis

@broker.subscriber(stream=StreamSub("stream", group="group", consumer="consumer"))
async def handler(msg: RedisStreamMessage, redis: Redis) -> None:
    ...

vs. the version that works:

from faststream.redis.annotations import Redis

Real usage

A worker running faststream[redis] in production hit this and left a warning to their future selves in the handler docstring — SW-Maestro-17th-HBB/Kkori-AI, worker/src/main.py#L186-L191:

redis 는 FastStream 이 Context 로 넣어주는 커넥션이다. redis.asyncio.Redis 를 그대로 힌트로 쓰면 주입되지 않고 검증 오류가 난다.

("redis is the connection FastStream injects via Context. If you use redis.asyncio.Redis directly as the hint, it is not injected and you get a validation error.")

They also import it aliased — from faststream.redis.annotations import Redis as InjectedRedis — which suggests the name collision cost them enough to want it visible at the call site.

Expected behavior

Two things would each be enough on their own:

1. Documentation — the Redis pages show Redis in examples without making the import path explicit enough to survive a copy-paste. A short note that the annotation comes from faststream.redis.annotations, and that the same-named driver class will not work, closes it.
2. A better failure — when an argument is annotated with a broker client class that has a matching FastStream annotation, say so in the error instead of failing validation on a missing message field.

Additional context

Found by reading a real user's source, not reported by them — they worked around it and moved on, which is the reason this kind of papercut stays invisible. Companion finding: #3064.

@IvanKirpichnikov — the docs half is cheap; the error-message half is your call.


#bug #documentation #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Bug: `StreamSub` warns that `no_ack` has no effect with a consumer group, but the flag is forwarded to XREADGROUP NOACK (#3126)


Describe the bug

StreamSub(..., group=..., consumer=..., no_ack=True) emits

RuntimeWarning: `no_ack` has no effect with consumer group

but the flag does have an effect: it reaches Redis as XREADGROUP ... NOACK, so entries never enter the PEL. The warning is a leftover from a time when the flag was not forwarded.

How to reproduce

from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()


@broker.subscriber(
    stream=StreamSub("my-stream", group="my-group", consumer="c1", no_ack=True),
)
async def handler(msg: str) -> None:
    ...

Constructing the StreamSub prints the warning. Running the app shows that the flag is honoured anyway: XPENDING my-stream my-group stays empty after messages are consumed.

Expected behavior

No warning. no_ack=True with a consumer group is a supported, documented mode ("equivalent to acknowledging the message when it is read", see docs/docs/en/redis/streams/groups.md).

Observed behavior

The warning is emitted in faststream/redis/schemas/stream_sub.py (the elif no_ack: branch under if group and consumer:), while the rest of the code base treats the flag as live:

• faststream/redis/subscriber/usecases/stream_subscriber.py passes noack=stream.no_ack to XREADGROUP;
• faststream/redis/subscriber/config.py switches the subscriber to AckPolicy.MANUAL when no_ack is set;
• faststream/redis/testing.py skips the PEL for no_ack subscribers;
• #3049 added a SetupError for claim_min_idle_time + no_ack, which only makes sense if no_ack works with groups.

History

• The warning was introduced in 0.3.5 (#1048), when no_ack was indeed not forwarded to XREADGROUP.
• #2309 kept the warning and reset no_ack = False right after it, so the flag really was ignored for a while.
• 0.6.0 (#1779) dropped that reset, and noack=stream.no_ack has been sent to Redis since, but the warning stayed.

The sibling warning, "no_ack is not supported by consumer group with last_id other than >", is a separate case and should be checked on its own: Redis documents NOACK for XREADGROUP regardless of the id, so it may be stale too.

Suggested fix

Remove the elif no_ack: warning branch in StreamSub.__init__ and add a test asserting that StreamSub("s", group="g", consumer="c", no_ack=True) raises no warning and that the subscriber calls xreadgroup with noack=True. Keep the claim_min_idle_time + no_ack SetupError as is.

Environment

Reproduced on main (commit 0307f09).


#bug #good_first_issue #redis #faststream #ag2ai
sent via relator