Slipstream

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

pip install slipstream-async

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

Slipstream components interoperate with basic python building blocks:

• Any-thing can be passed around as data
• Any Callable may be used as a sink
• AsyncIterables act as sources
• Parallelize through handle

A many-to-many relation is established by passing multiple sources / sinks.

Quickstart

Install Slipstream along with aiokafka (latest):

pip install slipstream-async[kafka]

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

docker compose up broker -d

Copy-paste this snippet.

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). Produce is fail-fast by default; set produce_retries / produce_retry_backoff to retry retriable broker errors (sources pause via Conf.signal_iterables while retrying)
• slipstream.codecs.JsonCodec: serialize and deserialize json messages
• slipstream.Checkpoint: recover from stream downtimes