riko 0.74.0

Why I built riko

I wanted a small-footprint, pure-Python library for processing data streams. I wanted to fetch RSS feeds and perform custom transformations without the complexities of distributed compute engines, workflow schedulers, clusters, or message queues.

riko, a primarily pull-based in-process library, is the result.

Why you should use riko

riko provides a number of benefits / differences from other stream processing applications:

• a small footprint (CPU and memory usage)

• native RSS/Atom support

• simple installation and usage

• a pure Python library supporting v3.12+

• built-in modular pipes to filter, sort, and modify streams

Why you shouldn’t use riko

riko is usually not the right tool when you need: distributed execution, durable scheduling, automatic retries, continual data monitoring, a workflow UI, event-triggered actions, or query optimization.

Choosing riko

The projects below overlap with riko in different ways. Some are embedded libraries, while others are distributed engines, workflow orchestrators, or data-integration platforms. RSS/Atom role distinguishes first-party feed support from functionality that requires a custom source, connector, tap, or task.

Choose riko when a pipeline should run directly inside a Python application without a separate scheduler, service, or cluster. It provides first-party RSS/Atom processing and supports synchronous, asynchronous, and thread-pooled local execution.

Choose RxPY for reactive event composition, Huginn for persistent UI-managed automation, and Flink, Storm, Spark, or Apache Beam when distributed execution is required. Luigi and Prefect orchestrate tasks, while Airbyte and Meltano focus on data-integration workflows. Meltano commonly runs Singer taps and targets and can add scheduling through an orchestration plugin. These tools may run riko as one step in a larger workflow rather than replace its in-process transformation API.

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
>>>
>>> items = [{'title': 'riko pt. 1'}, {'title': 'riko pt. 2'}]
>>> stream = SyncPipe('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
>>>
>>> items = [{'title': 'riko pt. 1'}]
>>> stream = SyncPipe('hash', items, field='title')
>>> next(stream)['hash']
1104819838

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

>>> from riko import SyncPipe
>>>
>>> items = [{'title': 'riko pt. 1'}]
>>> stream = SyncPipe('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
>>>
>>> items = [{'title': 'riko pt 1'}, {'title': 'riko pt 2'}]
>>> list(SyncPipe('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
>>>
>>> stream = SyncPipe('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 SyncPipe
>>>
>>> attrs = {'key': 'title', 'value': 'riko pt. 1'}
>>> next(SyncPipe('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
>>>
>>> metadata = get_module_metadata('fetchpage')
>>> metadata.name, metadata.type, metadata.subtype
('fetchpage', 'processor', 'source')
>>> metadata = get_module_metadata('count')
>>> metadata.name, metadata.type, metadata.subtype
('count', 'operator', 'aggregator')

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

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

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

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

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, SyncPipe
>>>
>>> stream = SyncPipe('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, 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('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, 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('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
>>>
>>> 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('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 51 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
>>>
>>> flow = SyncPipe('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.