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.5
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.5.tar.gz | 36.0 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| kstreams-0.34.5-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 81.6 kB
Release files / kstreams-0.34.5.tar.gz
| Download URL | kstreams-0.34.5.tar.gz |
|---|---|
| Size | 36.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
14fab7c04578a13d7d25ce3df0c1ca5e5b8fbdfd5a2aac925f1a90d6a986769a
|
|
BLAKE2b-256 checksum How to use checksums |
5030569bac346421a4dac97135128186f0af2fb711f956067f0c2f6078414033
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.5.1 CPython/3.14.7 Linux/6.17.0-1022-azure
|
Release files / kstreams-0.34.5-py3-none-any.whl
| Download URL | kstreams-0.34.5-py3-none-any.whl |
|---|---|
| Size | 45.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
b0ea501feb073a36574da104e457cc0dc69eb9a57ceb1d1999f97a6163c0651e
|
|
BLAKE2b-256 checksum How to use checksums |
25790e0ada80118eda2c886a613abe0a81d8b009edf6468af62e516e1e5f74a3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.5.1 CPython/3.14.7 Linux/6.17.0-1022-azure
|
