Effortless event stream integration for your services
FastStream simplifies the process of writing producers and consumers for message queues, handling all the parsing, networking and documentation generation automatically.
Making streaming microservices has never been easier. Designed with junior developers in mind, FastStream simplifies your work while keeping the door open for more advanced use cases. Here's a look at the core features that make FastStream a go-to framework for modern, data-centric microservices.
Multiple Brokers: FastStream provides a unified API to work across multiple message brokers (Kafka, RabbitMQ, NATS, Redis support)
Pydantic Validation: Leverage Pydantic's validation capabilities to serialize and validates incoming messages
Automatic Docs: Stay ahead with automatic AsyncAPI documentation
Intuitive: Full-typed editor support makes your development experience smooth, catching errors before they reach runtime
Powerful Dependency Injection System: Manage your service dependencies efficiently with FastStream's built-in DI system
Testable: Supports in-memory tests, making your CI/CD pipeline faster and more reliable
Extensible: Use extensions for lifespans, custom serialization and middleware
Integrations: FastStream is fully compatible with any HTTP framework you want (FastAPI especially)
Built for Automatic Code Generation: FastStream is optimized for automatic code generation using advanced models like GPT and Llama
That's FastStream in a nutshell—easy, efficient, and powerful. Whether you're just starting with streaming microservices or looking to scale, FastStream has got you covered.
Documentation: https://faststream.airt.ai/latest/
FastStream is a new package based on the ideas and experiences gained from FastKafka and Propan. By joining our forces, we picked up the best from both packages and created a unified way to write services capable of processing streamed data regardless of the underlying protocol. We'll continue to maintain both packages, but new development will be in this project. If you are starting a new service, this package is the recommended way to do it.
FastStream works on Linux, macOS, Windows and most Unix-style operating systems.
You can install it with pip as usual:
pip install faststream[kafka]
# or
pip install faststream[rabbit]
# or
pip install faststream[nats]
# or
pip install faststream[redis]By default FastStream uses PydanticV2 written in Rust, but you can downgrade it manually, if your platform has no Rust support - FastStream will work correctly with PydanticV1 as well.
FastStream brokers provide convenient function decorators @broker.subscriber
and @broker.publisher to allow you to delegate the actual process of:
consuming and producing data to Event queues, and
decoding and encoding JSON-encoded messages
These decorators make it easy to specify the processing logic for your consumers and producers, allowing you to focus on the core business logic of your application without worrying about the underlying integration.
Also, FastStream uses Pydantic to parse input JSON-encoded data into Python objects, making it easy to work with structured data in your applications, so you can serialize your input messages just using type annotations.
Here is an example Python app using FastStream that consumes data from an incoming data stream and outputs the data to another one:
fromfaststreamimportFastStreamfromfaststream.kafkaimportKafkaBroker# from faststream.rabbit import RabbitBroker# from faststream.nats import NatsBroker# from faststream.redis import RedisBrokerbroker=KafkaBroker("localhost:9092")
# broker = RabbitBroker("amqp://guest:guest@localhost:5672/")# broker = NatsBroker("nats://localhost:4222/")# broker = RedisBroker("redis://localhost:6379/")app=FastStream(broker)
@broker.subscriber("in")@broker.publisher("out")asyncdefhandle_msg(user: str, user_id: int) ->str:
returnf"User: {user_id} - {user} registered"Also, Pydantic’s BaseModel class allows you
to define messages using a declarative syntax, making it easy to specify the fields and types of your messages.
frompydanticimportBaseModel, Field, PositiveIntfromfaststreamimportFastStreamfromfaststream.kafkaimportKafkaBrokerbroker=KafkaBroker("localhost:9092")
app=FastStream(broker)
classUser(BaseModel):
user: str=Field(..., examples=["John"])
user_id: PositiveInt=Field(..., examples=["1"])
@broker.subscriber("in")@broker.publisher("out")asyncdefhandle_msg(data: User) ->str:
returnf"User: {data.user} - {data.user_id} registered"The service can be tested using the TestBroker context managers, which, by default, puts the Broker into "testing mode".
The Tester will redirect your subscriber and publisher decorated functions to the InMemory brokers, allowing you to quickly test your app without the need for a running broker and all its dependencies.
Using pytest, the test for our service would look like this:
# Code above omitted 👆importpytestimportpydanticfromfaststream.kafkaimportTestKafkaBroker@pytest.mark.asyncioasyncdeftest_correct():
asyncwithTestKafkaBroker(broker) asbr:
awaitbr.publish({
"user": "John",
"user_id": 1,
}, "in")
@pytest.mark.asyncioasyncdeftest_invalid():
asyncwithTestKafkaBroker(broker) asbr:
withpytest.raises(pydantic.ValidationError):
awaitbr.publish("wrong message", "in")The application can be started using built-in FastStream CLI command.
To run the service, use the FastStream CLI command and pass the module (in this case, the file where the app implementation is located) and the app symbol to the command.
faststream run basic:appAfter running the command, you should see the following output:
INFO - FastStream app starting...
INFO - input_data | - `HandleMsg` waiting for messages
INFO - FastStream app started successfully! To exit press CTRL+CAlso, FastStream provides you with a great hot reload feature to improve your Development Experience
faststream run basic:app --reloadAnd multiprocessing horizontal scaling feature as well:
faststream run basic:app --workers 3You can learn more about CLI features here
FastStream automatically generates documentation for your project according to the AsyncAPI specification. You can work with both generated artifacts and place a web view of your documentation on resources available to related teams.
The availability of such documentation significantly simplifies the integration of services: you can immediately see what channels and message formats the application works with. And most importantly, it won't cost anything - FastStream has already created the docs for you!
FastStream (thanks to FastDepends) has a dependency management system similar to pytest fixtures and FastAPI Depends at the same time. Function arguments declare which dependencies you want are needed, and a special decorator delivers them from the global Context object.
fromfaststreamimportDepends, Loggerasyncdefbase_dep(user_id: int) ->bool:
returnTrue@broker.subscriber("in-test")asyncdefbase_handler(user: str,
logger: Logger,
dep: bool=Depends(base_dep)):
assertdepisTruelogger.info(user)You can use FastStreamMQBrokers without a FastStream application.
Just start and stop them according to your application's lifespan.
fromaiohttpimportwebfromfaststream.kafkaimportKafkaBrokerbroker=KafkaBroker("localhost:9092")
@broker.subscriber("test")asyncdefbase_handler(body):
print(body)
asyncdefstart_broker(app):
awaitbroker.start()
asyncdefstop_broker(app):
awaitbroker.close()
asyncdefhello(request):
returnweb.Response(text="Hello, world")
app=web.Application()
app.add_routes([web.get("/", hello)])
app.on_startup.append(start_broker)
app.on_cleanup.append(stop_broker)
if__name__=="__main__":
web.run_app(app)Also, FastStream can be used as part of FastAPI.
Just import a StreamRouter you need and declare the message handler with the same @router.subscriber(...) and @router.publisher(...) decorators.
fromfastapiimportFastAPIfrompydanticimportBaseModelfromfaststream.kafka.fastapiimportKafkaRouterrouter=KafkaRouter("localhost:9092")
classIncoming(BaseModel):
m: dict@router.subscriber("test")@router.publisher("response")asyncdefhello(m: Incoming):
return {"response": "Hello, world!"}
app=FastAPI(lifespan=router.lifespan_context)
app.include_router(router)More integration features can be found here
As evident, FastStream is an incredibly user-friendly framework. However, we've taken it a step further and made it even more user-friendly! Introducing faststream-gen, a Python library that harnesses the power of generative AI to effortlessly generate FastStream applications. Simply describe your application requirements, and faststream-gen will generate a production-grade FastStream project that is ready to deploy in no time.
Save application description inside description.txt:
Create a FastStream application using localhost broker for testing and use the
default port number.
It should consume messages from the 'input_data' topic, where each message is a
JSON encoded object containing a single attribute: 'data'.
While consuming from the topic, increment the value of the data attribute by 1.
Finally, send message to the 'output_data' topic.
and run the following command to create a new FastStream project:
faststream_gen -i description.txt✨ Generating a new FastStream application!
✔ Application description validated.
✔ FastStream app skeleton code generated. akes around 15 to 45 seconds)...
✔ The app and the tests are generated. around 30 to 90 seconds)...
✔ New FastStream project created.
✔ Integration tests were successfully completed.
Tokens used: 10768
Total Cost (USD): $0.03284
✨ All files were successfully generated!We also invite you to explore our tutorial, where we will guide you through the process of utilizing the faststream-gen Python library to effortlessly create FastStream applications:
Please show your support and stay in touch by:
giving our GitHub repository a star, and
joining our Discord server
Your support helps us to stay in touch with you and encourages us to continue developing and improving the framework. Thank you for your support!
Thanks to all of these amazing people who made the project better!