FastStream Documentation

Redis Streams (introduced in Redis 5.0) are a data structure used for reliable and scalable data streaming, similar to Apache Kafka. They store an ordered sequence of messages, where each entry has a unique ID (including a timestamp) and key-value pairs.

Key features include:

• Persistence: Data is persisted and can be replayed by new consumers.
• Consumer Groups: Enables concurrent consumption and acknowledgment of entries by multiple consumers for partitioned processing.
• Range Queries: Allows querying data within specific ID ranges.

When using Redis Cluster, all stream operations (such as xadd, xreadgroup, xautoclaim, and xack) are automatically routed to the correct node by the RedisClusterBroker based on the key's hash slot.

FastStream is an asynchronous Python framework for building event-driven applications. Key features include:

• Multiple Brokers: Unified API for Kafka, RabbitMQ, NATS, and Redis.
• Built-in Serialization: Uses Pydantic or Msgspec for message validation and serialization.
• Automatic Docs: Generates AsyncAPI documentation automatically.
• Dependency Injection: Built-in DI system for managing service dependencies.
• Testable: Supports in-memory testing for faster CI/CD.
• Extensible: Supports extensions for lifespans, custom serialization, and middleware.
• Integrations: Fully compatible with HTTP frameworks like FastAPI.

FastStream's RabbitMQ support is built on top of aio-pika. It allows for complex routing scenarios using three main entities:

• Exchange: The entry point for messages from a publisher.
• Queue: The destination where messages are pushed to consumers.
• Binding: The relationship/link between a queue and an exchange, or between two exchanges.

Default Behavior: By default, all queues have a binding to the default exchange (Direct type) with a routing key that matches the queue's name. In FastStream, queues are connected to this exchange by default unless you explicitly specify another exchange.

To consume data in batches from a Kafka topic using FastStream, use the @broker.subscriber(...) decorator with the following configuration:

1. Set the batch parameter to True in the decorator.
2. Define the type hint for your message argument (msg) as a list of messages.

When configured this way, the subscriber will call your consuming function with a batch of messages collected from a single partition.

To gain full control over message acknowledgement, set ack_policy=AckPolicy.MANUAL in your subscriber. You can then access the KafkaMessage object via the handler arguments and call .ack() or .nack() directly.

• await msg.ack(): Acknowledges the message.
• await msg.nack(): Prevents offset commit, allowing the message to be consumed by another consumer in the same group.

If you manually acknowledge a message, FastStream will detect this and perform no further action at the end of the handler execution.

from faststream.confluent.annotations import KafkaMessage, AckPolicy

@broker.subscriber(
    "test", group_id="group", ack_policy=AckPolicy.MANUAL
)
async def base_handler(body: str, msg: KafkaMessage):
    await msg.ack()
    # or
    await msg.nack()

FastStream brokers use function decorators to handle data consumption, production, and serialization.

• @broker.subscriber(...): Decorates a function to consume data from an event queue.
• @broker.publisher(...): Decorates a function to produce data to an event queue.

By using type annotations, FastStream automatically handles the decoding and encoding of JSON-encoded messages into Python objects.

The FastAPI integration has been moved to a dedicated package. To use FastStream with FastAPI, install faststream_fastapi instead of using the built-in integration which is deprecated and will be removed in version 1.0.0.

pip install faststream_fastapi

You can connect to a Redis Cluster using a single URL, as the cluster client will automatically discover the remaining nodes. Alternatively, in multi-address environments, you can explicitly provide a list of seed nodes using the startup_nodes parameter.

Single URL connection

from faststream.redis import RedisClusterBroker

broker = RedisClusterBroker(url="redis://node1:7000")

Explicit seed nodes

broker = RedisClusterBroker(
    url="redis://node1:7000",
    startup_nodes=[
        ("node2", 7001),
        ("node3", 7002),
    ],
)

To implement load balancing where messages are delivered to only one consumer in a group, use MQTT shared subscriptions. In FastStream, you can enable this by passing the shared argument to the subscriber decorator.

Pass the group name as a string to shared. FastStream will automatically prepend $share/<group_name>/ to your topic filter. Multiple application instances using the same shared name and topic filter will compete for messages according to your broker's shared subscription implementation.