slipstream-async 0.0.0a0

Streamline your stream processing.

Slipstream

Slipstream provides a data-flow model to simplify development of stateful streaming applications.

pip install slipstream

from asyncio import run

from slipstream import handle, stream


async def messages():
    for emoji in '🏆📞🐟👌':
        yield emoji


@handle(messages(), sink=[print])
def handle_message(msg):
    yield f'Hello {msg}!'


if __name__ == '__main__':
    run(stream())

Hello 🏆!
Hello 📞!
Hello 🐟!
Hello 👌!

Usage

Async iterables are sources, (async) callables are sinks.

Decorate handler functions using handle, then run stream to start processing:

Multiple sources and sinks can be provided to establish many-to-many relations between them. The 4 emoji's were printed using the callable print.

Quickstart

Install aiokafka (latest) along with slipstream:

pip install slipstream[kafka]

Spin up a local Kafka broker with docker-compose.yml, using localhost:29091 to connect:

docker compose up broker -d

Follow the docs and set up a Kafka connection: slipstream.readthedocs.io.

Features

• slipstream.handle: bind streams (iterables) and sinks (callables) to user defined handler functions
• slipstream.stream: start streaming
• slipstream.Topic: consume from (iterable), and produce to (callable) kafka using aiokafka
• slipstream.Cache: store data to disk using rocksdict
• slipstream.Conf: set global kafka configuration (can be overridden per topic)
• slipstream.codecs.JsonCodec: serialize and deserialize json messages