riko stream processing for embedded Python pipelines

riko's local stream-processing scope

riko addresses the gap between ad hoc Python scripts and heavier stream-processing infrastructure. It is aimed at developers who want to fetch or load records, transform them, aggregate them, and observe results through a reusable pipeline without first deploying a scheduler, cluster, or message queue. The intended workloads include RSS and Atom feeds, web pages, text, and structured files. The README also positions riko as a library for pipelines represented in Python or JSON, as well as an embedded execution engine.

Its core abstraction is deliberately small: dictionary-like records move through pipes, and each pipe receives a configuration. A pipeline can filter, sort, parse, transform, aggregate, and compose those records. This makes the model useful for repeatable ingestion and transformation code that should remain close to an application rather than becoming a separately operated service.

The boundary matters more than the feature list. riko explicitly does not target distributed stream processing, durable workflow scheduling, recovery after worker failure, persistent retry management, task dependencies, a workflow user interface, or dataframe-scale columnar query optimization. A team looking for one of those capabilities should not treat the local execution options as substitutes for a stateful or distributed platform.

How records move through riko pipes

The repository describes an iterator-oriented model. Pipes are composed into a flow, and records emitted by one stage become inputs to the next. The quick-start example starts with `Sources.FETCHPAGE`, applies string replacement, tokenization, and counting, then retrieves emitted records with `next(flow)`. The README presents `next(flow)` returning records such as `{'Tidy': 1}` and `{'your': 1}`, which indicates a sequence of aggregate outputs rather than one final collection.

The configuration is part of the processing model. In the example, `fetch_conf` selects a URL, a start and end marker, and HTML detagging. The replacement configuration lists literal find and replace rules. Tokenization receives `{'delimiter': ' '}` and `emit=True`, while counting receives `{'count_key': 'content'}`. This keeps transformation details outside the pipeline's structural code, but it also means a flow's behavior depends heavily on configuration shape and emitted key names.

riko provides synchronous and asynchronous APIs plus local thread and process-pool execution. Based on the documentation, execution mode is a choice within an embedded pipeline rather than a distributed worker model. The JSON pipeline support and tools for inspecting, executing, and compiling pipelines extend the same abstraction to configuration-driven use, although the README does not provide a complete JSON example or compilation command.

Installing riko and running the word-count flow

riko requires Python 3.12 or later. Its packaging metadata classifies support for CPython 3.12, 3.13, and 3.14, and the README states that it is known to work with those versions. Install the latest published release from PyPI with the documented command:

python -m pip install riko

The default installation is described as a slim core. Optional dependency groups exist for performance-oriented packages, asynchronous support, and finance support. The README does not document the exact pip syntax for selecting those groups, so that syntax should be taken from the installation documentation rather than guessed.

After installation, the following complete example follows the README's quick start. It obtains a local fixture path through `get_path`, composes four processing stages, and requests the first emitted records.

from riko import get_path, Sources, SyncPipe

url = get_path('users.jyu.fi.html')
fetch_conf = {'url': url, 'start': '<body>', 'end': '</body>', 'detag': True}
replace_conf = {
    'rule': [{'find': '\r\n', 'replace': ' '}, {'find': '\n', 'replace': ' '}]
}

flow = (
    SyncPipe(Sources.FETCHPAGE, conf=fetch_conf)
    .strreplace(conf=replace_conf, assign='content')
    .tokenizer(conf={'delimiter': ' '}, emit=True)
    .count(conf={'count_key': 'content'})
)

next(flow)

The README example uses `next(flow)` directly and shows individual word records. The documented retrieval step returns an individual record. Its output depends on the fixture and processing settings, so the README example should be treated as the expected record shape, not a promised result for every input.

Choosing riko execution modes and pipe types

riko is not limited to the synchronous flow in the quick start. The library also exposes asynchronous execution and parallel execution through threads or processes. That choice can affect how a pipeline fits an application, but the README does not give enough detail to compare queue behavior, worker counts, serialization rules, cancellation, or exception propagation across all four modes. Those points require checking the relevant library and installation documentation before selecting a mode.

The pipe catalog is another selection criterion. riko's listed capabilities cover filtering, sorting, parsing, transforming, aggregating, and composing streams, with first-class support for RSS, Atom, and web-content processing. Its dependencies include `feedparser`, `requests`, `html5lib`, `networkx`, and text and date-processing packages, which are consistent with those documented domains. They do not establish support for databases, distributed brokers, or arbitrary external destinations, so this review does not infer such connectors.

The lazy iterator interface is useful when an application wants to consume records progressively. It can also complicate lifecycle management. The quick start must explicitly request an emitted record with `next(flow)`, and the README does not explain how to configure flow completion, large backlogs, retries, or cancellation. Applications that need those controls should inspect the pipe and execution documentation for the exact APIs rather than assuming iterator behavior supplies them.

Where riko is the wrong execution layer

The clearest failure mode is assigning distributed or durable work to a local embedded engine. riko does not claim durable keyed state, recovery after worker failure, persistent scheduling, retries, or dependency management. A pipeline that must retain progress through process loss therefore falls outside the documented design. The same boundary applies when execution must span a cluster or when operators need a workflow service and UI.

riko is also a weak fit for analytical workloads centered on large columnar tables. The README distinguishes it from dataframe query engines and does not claim query optimization at that scale. Its records move through configured pipes, which is a different design from relational reshaping, lazy table algebra, or dataframe operations. A project whose main requirement is joins, table reshaping, and data-quality transformations across table-shaped data may find `petl` more directly aligned.

Another boundary is source-to-destination replication. riko focuses on processing and composing record streams. If the primary task is moving data from many standardized connectors into warehouses, lakes, or databases, `dlt` emphasizes schema-aware ingestion and structured destinations, while `Singer` emphasizes standardized taps, targets, records, schemas, and replication state. These are different contracts, not merely different syntaxes.

riko alternatives, licensing, and maintenance

The README compares several alternatives. `dlt` is oriented toward extracting data from REST APIs and loading it into structured destinations, with pagination, authentication, and schema normalization. `Singer` is a connector protocol for data replication. `Bytewax` focuses on stateful streaming runtimes, keyed state, recovery, workers, and distributed processing. `Bonobo` provides injectable services and I/O for traditional ETL graphs, while `Streamz` emphasizes continuous push-oriented graphs, branching, backpressure, and live windows. `petl` centers on lazy table algebra for joins, reshaping, and data-quality work. riko's stated distinction is its broad set of reusable, JSON-serializable, configuration-driven pipes.

riko is distributed under the MIT license. The project metadata includes the `LICENSE` file, but this review does not interpret the legal effect of that license or provide legal advice. The practical licensing point is that the repository declares MIT rather than placing the code under a copyleft license.

The repository is not archived, and its last push was on 2026-10-16. Recent releases include v0.77.3 on 2026-09-01, v0.77.2 on 2026-08-31, and v0.77.1 on 2026-08-31. This is evidence of recent repository activity, but the repository does not define an upgrade policy, compatibility policy across releases, or migration procedure. Because the default version in `pyproject.toml` is v0.77.3, an upgrade review should inspect changes in the target release rather than assume that configuration and pipe behavior remain unchanged.

Editorial conclusion

Python developers should consider riko when records can flow through an embedded, configuration-driven pipeline without cluster infrastructure, durable state, or a scheduler. It is a poor fit for distributed execution, recoverable keyed workflows, or dataframe-scale analytics. Before adoption, verify first that your records fit the dictionary-like pipe model, that the required sync, async, thread, or process behavior is available, and that no unconfirmed operational requirement is being assigned to riko that its own documentation places in another system.

Frequently asked questions

Official sources

1. Issues
2. License: MIT
3. nerevu/riko on GitHub
4. README
5. Releases