FastStream

0.6.7 · active · verified Sun Apr 12

FastStream is a modern, asynchronous framework designed to simplify interactions with various messaging queues (Kafka, RabbitMQ, Redis, NATS, MQTT). It provides an intuitive, FastAPI-like API for building message-driven applications, handling message serialization, dependency injection, and health checks. The current stable version is 0.6.7, with frequent minor releases and a significant 0.7.0 release candidate bringing breaking changes.

Warnings

breaking Version 0.7.0 (release candidate available) removes several deprecated features. Specifically: publisher/subscriber-level middlewares, `ack_policy` now replaces several deprecated options, `RedisJSONMessageParser` is removed (Redis services must now use binary message format), and `broker.close` is replaced by `broker.stop`.
Fix: Review your application for usage of deprecated features. For Redis, ensure your message formats are binary. Replace `broker.close()` with `broker.stop()`.
breaking As of 0.7.0rc0 (and likely stable 0.7.0), import paths for specific brokers have changed. They are now directly under `faststream.<broker_name>` (e.g., `from faststream.kafka import KafkaBroker`) instead of `faststream.broker.<broker_name>`.
Fix: Update all broker import statements to the new direct path, e.g., `from faststream.kafka import KafkaBroker`.
gotcha FastStream's core installation (`pip install faststream`) does not include broker-specific client libraries. You must install FastStream with the appropriate extras, e.g., `pip install "faststream[kafka]"`, `"faststream[redis]"`, `"faststream[mqtt]"`, etc. Failing to do so will result in `ModuleNotFoundError` when trying to import or use a broker.
Fix: Always install FastStream with the necessary broker extras for your chosen messaging queue (e.g., `pip install "faststream[kafka,redis]"` for both).
breaking Version 0.6.0 introduced significant changes to the Middleware API and Router API. Applications upgrading from versions prior to 0.6.0 may experience breaking changes related to how middlewares are defined and applied, and how routers are configured.
Fix: Consult the official FastStream 0.6.0 release notes and documentation regarding the new Middleware and Router APIs. Adjust your code to conform to the finalized API designs.

Install

pip install "faststream[kafka]" Install with Kafka support
pip install faststream Install core FastStream (no broker)

Imports

FastStream
```
from faststream import FastStream
```
Logger
```
from faststream import Logger
```
KafkaBroker
```
from faststream.kafka import KafkaBroker
```
Broker imports are now directly from `faststream.<broker_name>` as of v0.7.0rc0 and likely stable 0.7.0. Older versions used `faststream.broker.<broker_name>`.
MQTTBroker
```
from faststream.mqtt import MQTTBroker
```
Broker imports are now directly from `faststream.<broker_name>` as of v0.7.0rc0 and likely stable 0.7.0. Older versions used `faststream.broker.<broker_name>`.
ContextRef
```
from faststream.types import ContextRef
```

Quickstart

This quickstart demonstrates a basic FastStream application with Kafka. It defines a Kafka broker, an application instance, and a subscriber that listens to 'test-topic'. A message is published on application startup for immediate testing. Ensure Kafka is running and FastStream is installed with Kafka extras to run.

import os
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker

# Configure Kafka broker connection
# Use 'localhost:9092' for local Kafka, or environment variable for production
KAFKA_BROKER_URL = os.environ.get('KAFKA_BROKER_URL', 'localhost:9092')

broker = KafkaBroker(KAFKA_BROKER_URL)
app = FastStream(broker)

@broker.subscriber('test-topic')
async def handle_message(msg: str, logger: Logger):
    logger.info(f"Received message: {msg}")

@app.on_startup
async def app_startup():
    logger = Logger().bind(service="startup")
    logger.info("FastStream application started")
    # Example: publish a message on startup
    await broker.publish("Hello, FastStream!", topic='test-topic')

@app.on_shutdown
async def app_shutdown():
    logger = Logger().bind(service="shutdown")
    logger.info("FastStream application stopped")

# To run this application:
# 1. Ensure Kafka is running.
# 2. Save as `app.py`.
# 3. `pip install "faststream[kafka]"`
# 4. `faststream run app:app`

view raw JSON →