Skip to content

QUICK START#

Install using pip:

pip install "faststream[kafka]"

Tip

To start a new project, we need a test broker container 

docker run -d --rm -p 9092:9092 --name test-mq \
-e KAFKA_NODE_ID=1 \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_BROKER_ID=1 \
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
-e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk \
confluentinc/cp-kafka:8.0.0


pip install "faststream[confluent]"

Tip

To start a new project, we need a test broker container 

docker run -d --rm -p 9092:9092 --name test-mq \
-e KAFKA_NODE_ID=1 \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_BROKER_ID=1 \
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
-e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk \
confluentinc/cp-kafka:8.0.0


pip install "faststream[rabbit]"

Tip

To start a new project, we need a test broker container 

docker run -d --rm -p 5672:5672 --name test-mq rabbitmq:alpine


pip install "faststream[nats]"

Tip

To start a new project, we need a test broker container 

docker run -d --rm -p 4222:4222 --name test-mq nats -js


pip install "faststream[redis]"

Tip

To start a new project, we need a test broker container 

docker run -d --rm -p 6379:6379 --name test-mq redis

Basic Usage#

Note

Before continuing with the next steps, make sure you install Fastream CLI. 

pip install "faststream[cli]"

To create a basic application, add the following code to a new file (e.g. serve.py):

serve.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")

app = FastStream(broker)


@broker.subscriber("test")
async def base_handler(body):
    print(body)
serve.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream import FastStream
from faststream.confluent import KafkaBroker

broker = KafkaBroker("localhost:9092")

app = FastStream(broker)


@broker.subscriber("test")
async def base_handler(body):
    print(body)
serve.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream import FastStream
from faststream.rabbit import RabbitBroker

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")

app = FastStream(broker)


@broker.subscriber("test")
async def base_handler(body):
    print(body)
serve.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream import FastStream
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")

app = FastStream(broker)


@broker.subscriber("test")
async def base_handler(body):
    print(body)
serve.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from faststream import FastStream
from faststream.redis import RedisBroker

broker = RedisBroker("redis://localhost:6379")

app = FastStream(broker)


@broker.subscriber("test")
async def base_handler(body):
    print(body)

And just run this command:

faststream run serve:app

After running the command, you should see the following output:

INFO     - FastStream app starting...
INFO     - test |            - `BaseHandler` waiting for messages
INFO     - FastStream app started successfully! To exit, press CTRL+C

Enjoy your new development experience!

Manual run#

Also, you can run the FastStream application manually, as a regular async function:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import asyncio

from faststream import FastStream
from faststream.kafka import KafkaBroker


async def main():
    broker = KafkaBroker("localhost:9092")
    app = FastStream(broker)
    await app.run()  # blocking method

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import asyncio

from faststream import FastStream
from faststream.confluent import KafkaBroker


async def main():
    broker = KafkaBroker("localhost:9092")
    app = FastStream(broker)
    await app.run()  # blocking method

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import asyncio

from faststream import FastStream
from faststream.rabbit import RabbitBroker


async def main():
    broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
    app = FastStream(broker)
    await app.run()  # blocking method

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import asyncio

from faststream import FastStream
from faststream.nats import NatsBroker


async def main():
    broker = NatsBroker("nats://localhost:4222")
    app = FastStream(broker)
    await app.run()  # blocking method

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import asyncio

from faststream import FastStream
from faststream.redis import RedisBroker


async def main():
    broker = RedisBroker("redis://localhost:6379")
    app = FastStream(broker)
    await app.run()  # blocking method

if __name__ == "__main__":
    asyncio.run(main())

Other tools integrations#

If you want to use FastStream as part of another framework service, you probably don't need to utilize the FastStream object at all, as it is primarily intended as a CLI tool target. Instead, you can start and stop your broker as part of another framework's lifespan. You can find such examples in the integrations section.

Don't forget to stop the test broker container 
docker container stop test-mq