2025 Python Packaging Survey is now live!  Take the survey now

Build simple kafka streams applications

Navigation

Verified details

These details have been verified by PyPI
Maintainers
Avatar for gmic from gravatar.com gmic Avatar for marcosschroh from gravatar.com marcosschroh Avatar for Willemoes from gravatar.com Willemoes

Unverified details

These details have not been verified by PyPI
Meta
  • License: Apache Software License (Apache-2.0)
  • Author: Marcos Schroh
  • Tags stream , processing , kafka , event streaming
  • Requires: Python >=3.8, <4.0
Classifiers
Report project as malware

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.

Build status codecov python version

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
  • 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 (black and isort)

./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

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page