Build simple kafka streams applications
Project description
Kstreams
kstreams
is a library/micro framework to use with kafka
. It has simple kafka streams implementation that gives certain guarantees, see below.
Documentation: https://kpn.github.io/kstreams/
Installation
pip install kstreams
You will need a worker, we recommend aiorun
pip install aiorun
Usage
import aiorun
from kstreams import create_engine, ConsumerRecord
stream_engine = create_engine(title="my-stream-engine")
@stream_engine.stream("local--kstream")
async def consume(cr: ConsumerRecord):
print(f"Event consumed: headers: {cr.headers}, payload: {cr.value}")
async def produce():
payload = b'{"message": "Hello world!"}'
for i in range(5):
metadata = await stream_engine.send("local--kstreams", value=payload)
print(f"Message sent: {metadata}")
async def start():
await stream_engine.start()
await produce()
async def shutdown(loop):
await stream_engine.stop()
if __name__ == "__main__":
aiorun.run(start(), stop_on_unhandled_errors=True, shutdown_callback=shutdown)
Features
- Produce events
- Consumer events with
Streams
-
Prometheus
metrics and custom monitoring - TestClient
- Custom Serialization and Deserialization
- Easy to integrate with any
async
framework. No tied to any library!! - Yield events from streams
- Opentelemetry Instrumentation
- Middlewares
- Hooks (on_startup, on_stop, after_startup, after_stop)
- Store (kafka streams pattern)
- Stream Join
- Windowing
Development
This repo requires the use of poetry instead of pip.
Note: If you want to have the virtualenv
in the same path as the project first you should run poetry config --local virtualenvs.in-project true
To install the dependencies just execute:
poetry install
Then you can activate the virtualenv
with
poetry shell
Run test:
./scripts/test
Run code formatting with ruff:
./scripts/format
Commit messages
We use conventional commits for the commit message.
The use of commitizen is recommended. Commitizen is part of the dev dependencies.
cz commit
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for kstreams-0.20.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0d9d5547e09e9712eb7a3fd1a4541614dadbe4317d95dccb6df6951440243e06 |
|
MD5 | d41fc3321f247953aadea72d06e53605 |
|
BLAKE2b-256 | 343290efa8fca1a8550f5ec6ee34eb30c3871e7d044ac3b892e5e81b4e31454b |