riko 0.76.1

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.

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.

Overview

Here’s the riko vocabulary at a glance:

Core concepts

The primary data structures in riko are the item and stream. An item is just a Python dictionary, and a stream is an iterator of item. You can create a stream manually with something as simple as iter([{'content': 'hello world'}]). You manipulate streams in riko via pipes. A pipe is simply a function that accepts either a stream or item, and returns a stream.

Through SyncPipe and AsyncPipe classes, pipes are composable: the output of each pipe is the input to the next pipe.

riko pipes come in three types: processor, operator, and splitter. An operator operates on a stream and is unable to handle individual items. E.g., count, filter, and reverse.

>>> from riko import SyncPipe, Transforms
>>>
>>> items = [{'title': 'riko pt. 1'}, {'title': 'riko pt. 2'}]
>>> stream = SyncPipe(Transforms.REVERSE, items)
>>> next(stream)
{'title': 'riko pt. 2'}

A processor processes an individual item and can be parallelized across threads or processes. E.g., fetchsitefeed, hash, itembuilder, and regex.

>>> from riko import SyncPipe, Transforms
>>>
>>> items = [{'title': 'riko pt. 1'}]
>>> stream = SyncPipe(Transforms.HASH, items, field='title')
>>> next(stream)['hash']
1104819838

Some processors, e.g., tokenizer, return multiple results.

>>> from riko import SyncPipe, Transforms
>>>
>>> items = [{'title': 'riko pt. 1'}]
>>> stream = SyncPipe(Transforms.TOKENIZER, items, conf={'delimiter': ' '}, field='title')
>>> list(stream)
[{'content': 'riko'}, {'content': 'pt.'}, {'content': '1'}]

operators are split into sub-types: aggregator and composer. aggregators, e.g., count, combine all items of an input stream into a new stream with a single item; while composers, e.g., filter, create a new stream containing some or all items of an input stream.

>>> from riko import SyncPipe, Transforms
>>>
>>> items = [{'title': 'riko pt 1'}, {'title': 'riko pt 2'}]
>>> list(SyncPipe(Transforms.COUNT, items))
[{'count': 2}]

Astute observers may have noticed from the “Word Count” example up top, that count can return multiple items if you pass in the count_key config option.

>>> from riko import SyncPipe, Transforms
>>>
>>> stream = SyncPipe(Transforms.COUNT, items, conf={'count_key': 'title'})
>>> list(stream)
[{'riko pt 1': 1}, {'riko pt 2': 1}]

processors are parallelizable and split into sub-types of source and transformer. A source, e.g., itembuilder, can create a stream, while a transformer, e.g. hash can only transform a source item.

>>> from riko import Sources, SyncPipe
>>>
>>> attrs = {'key': 'title', 'value': 'riko pt. 1'}
>>> next(SyncPipe(Sources.ITEMBUILDER, conf={'attrs': attrs}))
{'title': 'riko pt. 1'}

The following table summarizes these observations:

Note: Since some pipes support more than one subtype depending on their options, view the FAQ for steps on runtime discovery via discovering modules.

If you are unsure of the type of pipe you have, check its metadata.

>>> from riko import get_module_metadata, Sources, Transforms
>>>
>>> metadata = get_module_metadata(Sources.FETCHPAGE)
>>> metadata.name, metadata.type, metadata.subtype
('fetchpage', 'processor', 'source')
>>> metadata = get_module_metadata(Transforms.COUNT)
>>> metadata.name, metadata.type, metadata.subtype
('count', 'operator', 'aggregator')

Note: type and subtype are mutually exclusive: a subtype implies its type.

SyncPipe/AsyncPipe perform this check for you to allow for convenient method chaining and transparent parallelization.

>>> from riko import Sources, SyncPipe
>>>
>>> attrs = [
...     {'key': 'title', 'value': 'riko pt. 1'},
...     {'key': 'content', 'value': "Let's talk about riko!"}
... ]
>>> flow = SyncPipe(Sources.ITEMBUILDER, conf={'attrs': attrs}).hash()
>>> item = next(flow)
>>> item['title'], item['content'], item['hash']
('riko pt. 1', "Let's talk about riko!", 197222720)

The | operator chains the same way. It takes a module name or a (name, conf) tuple. The later is handy when the next pipe’s name is computed. A name may be a plain string or a member of the typed discovery tree (Sources/Transforms/ Sinks).

>>> from riko import Sources, SyncPipe, Transforms
>>>
>>> attrs = [
...     {'key': 'title', 'value': 'riko pt. 1'},
...     {'key': 'content', 'value': "Let's talk about riko!"}
... ]
>>> conf = {'attrs': attrs}
>>> item = next(SyncPipe(Sources.ITEMBUILDER, conf=conf) | Transforms.HASH)
>>> item['title'], item['hash']
('riko pt. 1', 197222720)

View the Cookbook for advanced examples including how to wire in values from other pipes or accept user input.

Usage Index

• Fetching data

• Synchronous processing

• Parallel processing

• Asynchronous processing

• Built-in pipes

• Pipeline lifecycle

Fetching data

riko can fetch data such as HTML, JSON, CSV, etc. from both local and remote filepaths via source pipes:

>>> from riko import get_path, Sources, SyncPipe
>>>
>>> stream = SyncPipe(Sources.FETCH, conf={'url': get_path('feed.xml')})
>>> item = next(stream)
>>> {'author', 'content', 'id', 'link', 'published', 'summary', 'title'} <= set(item)
True
>>> item['title'], item['author'], item['id']
('Donations', {'name': 'WriteToReply', 'uri': None}, 'http://writetoreply.org/?page_id=111')

View the FAQ for a complete list of supported file types and protocols; and Fetching data and feeds for more examples.

Synchronous processing

riko can modify a stream via transformer, composer, and aggregator pipes:

>>> from riko import get_path, Sources, SyncPipe
>>>
>>> fetch_conf = {'url': get_path('feed.xml')}
>>> filter_rule = {'field': 'title', 'op': 'contains', 'value': 'a'}
>>>
>>> # The following flow will:
>>> #   1. fetch a (cached) RSS feed
>>> #   2. filter for items with an 'a' in the title
>>> #   3. sort the items ascending by title
>>> #
>>> # Note: sorting is not lazy so take caution when using this pipe
>>>
>>> flow = (
...     SyncPipe(Sources.FETCH, conf=fetch_conf)   # 1
...     .filter(conf={'rule': filter_rule})        # 2
...     .sort(conf={'rule': {'field': 'title'}})   # 3
... )
>>>
>>> next(flow)['title']
'Donations'

View pipes for a complete list of available pipes.

Parallel processing

An example using riko’s parallel API to spawn a ThreadPool [1]

>>> from riko import get_path, Sources, SyncPipe
>>>
>>> fetch_conf = {'url': get_path('feed.xml')}
>>> filter_rule = {'field': 'title', 'op': 'contains', 'value': 'a'}
>>>
>>> # The following flow will:
>>> #   1. fetch a (cached) RSS feed
>>> #   2. filter for items with an 'a' in the title, in parallel (4 workers)
>>> #
>>> # Note: no point in sorting after the filter since parallel fetching doesn't
>>> # guarantee order
>>> flow = (
...     SyncPipe(Sources.FETCH, conf=fetch_conf, parallel=True, workers=4)  # 1
...     .filter(conf={'rule': filter_rule})                           # 2
... )
>>>
>>> sorted(item['title'] for item in flow)[:3]
['Donations', 'FAQ', 'General Comments']

Notes

Asynchronous processing

To enable asynchronous processing, you must install the async extra.

python -m pip install "riko[async]"

>>> from riko import AsyncPipe, get_path, issync, run, Sources
>>>
>>> fetch_conf = {'url': get_path('feed.xml')}
>>> filter_rule = {'field': 'title', 'op': 'contains', 'value': 'a'}
>>>
>>> # The following flow will:
>>> #   1. fetch a (cached) RSS feed
>>> #   2. filter for items with an 'a' in the title
>>>
>>> async def main():
...     stream = await (
...         AsyncPipe(Sources.FETCH, conf=fetch_conf)           # 1
...             .filter(conf={'rule': filter_rule}))            # 2
...
...     print(next(stream)['title'])
>>>
>>> print('Donations') if issync else run(main)
Donations

Built-in pipes

riko ships 52 built-in pipes. The table below summarizes them.

Pipeline lifecycle

SyncPipe/AsyncPipe represent a single execution: iterating one consumes the stream, and iterating it again yields an empty stream. Read the state/exhausted/closed/failed properties to inspect a pipe. Use it as a context manager (or call close()/terminate()) to release a parallel pipe’s worker pool deterministically.

>>> from riko import SyncPipe, Transforms
>>>
>>> flow = SyncPipe(Transforms.HASH, source=[{'content': 'a'}, {'content': 'b'}])
>>> flow.state
<PipeState.NEW: 'new'>
>>> len(list(flow))
2
>>> flow.state
<PipeState.EXHAUSTED: 'exhausted'>
>>> flow.exhausted
True

See the Cookbook for pool cleanup and the full state model.