riko: composable stream processing for Python

|ci| |pypi| |versions| |license|

.. contents:: :local: :depth: 2

Introduction

riko is a pure Python library_ for building data-processing streams. riko combines reusable, configuration-driven modular pipes_ with synchronous, asynchronous, and parallel execution_ APIs. It is particularly useful for processing RSS feeds, web content, text, and structured files.

riko also supplies a command-line interface_ for executing flows, i.e., stream processors aka pipelines.

Requirements & Installation

riko has been tested and is known to work on Python 3.12, 3.13, and 3.14.

Install the latest published release from PyPI:

.. code-block:: bash

python -m pip install riko

riko installs a slim core by default. View the installation doc_ for advanced installation options.

Quick start

The following example fetches a webpage, splits its text into words, and counts the number of times each word appears.

.. code-block:: python

>>> from riko import get_path, Sources, SyncPipe
>>>
>>> ### Set the pipe configurations ###
>>> #
>>> # Notes:
>>> #   1. look up cached html file in the `data` directory
>>> #   2. fetch text in the 'body' tag and strip html tags
>>> #   3. replace newlines with spaces and assign the result to 'content'
>>> #   4. split text in words using whitespace as the delimiter
>>> #   5. count the number of times each word appears
>>>
>>> url = get_path('users.jyu.fi.html')                   # 1
>>> fetch_conf = {'url': url, 'start': '', 'end': '', 'detag': True}
>>> replace_conf = {
...     'rule': [{'find': '\r\n', 'replace': ' '}, {'find': '\n', 'replace': ' '}]
... }
>>>
>>> flow = (
...     SyncPipe(Sources.FETCHPAGE, conf=fetch_conf)      # 2
...     .strreplace(conf=replace_conf, assign='content')  # 3
...     .tokenizer(conf={'delimiter': ' '}, emit=True)    # 4
...     .count(conf={'count_key': 'content'})             # 5
... )
>>>
>>> next(flow)
{'Tidy': 1}
>>> next(flow)
{'your': 1}

Motivation

Why I built riko ^^^^^^^^^^^^^^^^

I wanted a small-footprint, pure-Python library for processing data streams. In particular, I wanted to fetch RSS feeds and web pages and process records without needing to deploy a scheduler, cluster, or message queue.

The basic idea is deliberately simple: dictionary-like records flow through configurable pipes. Pipelines can run synchronously, asynchronous via async/await, or parallelized across threads or processes.

Why you should use riko ^^^^^^^^^^^^^^^^^^^^^^^

riko is a good fit when you want a batteries included, reusable, data-processing abstraction.

In particular, riko provides:

• a pure-Python, embedded execution model with no required external services
• a library of configuration-driven pipes for filtering, sorting, parsing, transforming, aggregating, and composing streams
• first-class RSS/Atom and web-content processing
• synchronous and asynchronous APIs
• local thread and process-pool execution
• lazy iterator-oriented processing
• simple Python or JSON pipeline configuration and definition
• tools to inspect, execute, and compile pipelines

Why you shouldn't use riko ^^^^^^^^^^^^^^^^^^^^^^^^^^

riko does not try to be a distributed stream-processing engine, durable workflow scheduler, or dataframe query engine.

It is usually not the right tool when you need:

• execution across a cluster
• durable keyed state and recovery after worker failure
• persistent scheduling, retries, or task dependency management
• a workflow service/UI
• event-triggered infrastructure automation
• dataframe-scale columnar analytics or query optimization

riko can instead run inside a worker or task managed by such systems.

Choosing riko ^^^^^^^^^^^^^

Several Python projects overlap with riko, but they optimize for different parts of the data-processing problem.

+------------------------------------------+------------------------------------------------------------------+ | Project | Distinctive strength | Prefer it for... | +==========+===============================+==================================================================+ | dlt_ | Declarative, schema-aware | moving data from REST APIs into warehouses/lakes/databases | | | ingestion | | +------------------------------------------+------------------------------------------------------------------+ | Singer_ | Standardized taps and targets | replicating data from various sources into many destinations | +------------------------------------------+------------------------------------------------------------------+ | Bytewax_ | Stateful streaming runtime | keyed state, recovery, workers, or distributed stream processing | +------------------------------------------+------------------------------------------------------------------+ | Bonobo_ | Injectable services and I/O | traditional ETL graphs and runtime-injected infrastructure | +------------------------------------------+------------------------------------------------------------------+ | Streamz_ | Continuous stream graphs | push-oriented streams, branching, backpressure, or live windows | +------------------------------------------+------------------------------------------------------------------+ | petl_ | Rich lazy table algebra | joins, reshaping, and data-quality operations | +------------------------------------------+------------------------------------------------------------------+ | riko | Config-driven pipelines | broad library of reusable, JSON serializable pipes | +------------------------------------------+------------------------------------------------------------------+

The closest comparison depends on what part of riko you care about.

dlt_ is a Python ingestion framework/library with similarities to riko in REST ingestion, incremental extraction, schema-aware loading, and Python-native data handling. dlt primarily allows you to "get data out a source reliably and into a well-structured destination." It provides primitives for pagination, auth, and schema normalization. This contrasts with riko's main use-case of processing and composing streams of records.

Singer_ is a connector protocol that standardizes how sources (taps) and destinations (targets) exchange records, schemas, and replication state. It overlaps with riko at the extraction and data-movement boundaries. Singer is a better fit when the primary goal is source-to-destination replication. riko instead places more emphasis on transforming and composing records.

Bytewax_ is the natural direction when a workload grows beyond riko's intended scope and requires durable keyed state, recovery, or distributed stream processing.

petl_ and Bonobo_, like riko, are both lightweight ETL libraries. Compared to riko, petl is more table-oriented and provides a deeper relational/data-wrangling vocabulary. Bonobo centers execution around an ETL graph of transformation nodes.

Streamz_ overlaps most with riko's stream-composition and fan-out model, but places more emphasis on continuous push-based streams, windowing, and reactive dataflow.

riko provides more "batteries included" data-processing vocabulary. It exposes common operations (filtering, truncating, searching, etc.) as configurable, reusable pipes rather than requiring a Python callable. riko also provides first-class support for web-content (RSS/Atom feeds, HTML/XML, and JSON) and a simple JSON-based pipeline definition format.

Design Principles

Overview ^^^^^^^^

Here's the riko vocabulary at a glance:

+---------------------+---------------------------------------+--------------------------------------------------+ | Term | Meaning | Example | +=====================+=======================================+==================================================+ | item | one dictionary-like record | {'title': 'Example'} | +---------------------+---------------------------------------+--------------------------------------------------+ | stream | an iterator of item | iter([{'title': 'Example'}]) or SyncPipe | +---------------------+---------------------------------------+--------------------------------------------------+ | pipe | a configured stream operation | join, ``slugify