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 - Subscribe to topics by
pattern -
Prometheusmetrics and custom monitoring - TestClient
- Custom Serialization and Deserialization
- Easy to integrate with any
asyncframework. 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
3 maintainers
Metadata
3 maintainers
Release files for kstreams 0.34.4
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| kstreams-0.34.4.tar.gz | 36.0 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| kstreams-0.34.4-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 81.5 kB
Release files / kstreams-0.34.4.tar.gz
| Download URL | kstreams-0.34.4.tar.gz |
|---|---|
| Size | 36.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
d0b8a6d46fc0e6b11c0a53539e28b5b1d9ac028f93741c854489ec722bfd69d0
|
|
BLAKE2b-256 checksum How to use checksums |
eb3ec7a445e66f9f851672f085a6f55a35d391eeb15374361f91c2651c68f85f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.4.3 CPython/3.14.7 Linux/6.17.0-1022-azure
|
Release files / kstreams-0.34.4-py3-none-any.whl
| Download URL | kstreams-0.34.4-py3-none-any.whl |
|---|---|
| Size | 45.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
47d2ab7e944551e2d2fedacdbe2712c7f8688832c6ff336a4265ce8859294a74
|
|
BLAKE2b-256 checksum How to use checksums |
a718aea0230d259ad2cccf2f730391e3a853d43e533322d450921d4810f97eec
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.4.3 CPython/3.14.7 Linux/6.17.0-1022-azure
|
