riko 0.73.1

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+

• builtin 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 below projects 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 Singer move data between systems. These tools may complement riko rather than replace it.

Overview

Here’s the riko vocabulary at a glance:

Notes:

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

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 items. 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. pipes are composable: the output of one pipe can be the input to another pipe.

riko pipes come in two flavors; operator and processor. An operator operates on an entire stream at once and is unable to handle individual items. E.g., count, filter, and reverse.

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

Some processor’s, e.g., tokenizer, return multiple results.

>>> from riko.modules.tokenizer import pipe
>>>
>>> item = {'title': 'riko pt. 1'}
>>> tokenizer_conf = {'delimiter': ' '}
>>> stream = pipe(item, conf=tokenizer_conf, field='title')
>>> list(stream)
[{'content': 'riko'}, {'content': 'pt.'}, {'content': '1'}]

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

>>> from riko.modules.count import pipe
>>>
>>> items = [{'title': 'riko pt 1'}, {'title': 'riko pt 2'}]
>>> list(pipe(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.

>>> stream = pipe(items, conf={'count_key': 'title'})
>>> list(stream)
[{'riko pt 1': 1}, {'riko pt 2': 1}]

processor’s 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.modules.itembuilder import pipe
>>>
>>> attrs = {'key': 'title', 'value': 'riko pt. 1'}
>>> next(pipe(conf={'attrs': attrs}))
{'title': 'riko pt. 1'}

The following table summaries these observations:

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

>>> from riko.modules import fetchpage, count
>>>
>>> fetchpage.pipe.name, fetchpage.pipe.type, fetchpage.pipe.subtype
('fetchpage', 'processor', 'source')
>>> count.pipe.name, count.pipe.type, count.pipe.subtype
('count', 'operator', 'aggregator')

The SyncPipe and AsyncPipe classes (among other things) 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'], isinstance(item['hash'], int)
('riko pt. 1', "Let's talk about riko!", True)

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

Notes:

• 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 pipe’s:

>>> from riko import get_path
>>> from riko.modules.fetch import pipe as fetch
>>> from riko.modules.fetchsitefeed import pipe as fetchsitefeed
>>>
>>> stream = 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 pipe’s:

>>> 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 alternate workflow creation for an alternative (function based) method for creating a stream. 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]                        # 3
['Donations', 'FAQ', 'General Comments']

Notes

Asynchronous processing

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

pip install riko[async]

>>> from riko import get_path, AsyncPipe
>>> from riko.bado import run, issync
>>>
>>> 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 below tables summarizes them. The FAQ has the complete per-module table.

Pipeline lifecycle

SyncPipe/AsyncPipe represent a single execution: iterating it consumes the stream, and iterating again yields an empty stream rather than silently re-running. 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.exhausted
True

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

CLI Usage

usage: run-pipe [pipeid]

description: Runs a riko pipe

positional arguments:

pipeid The pipe to run (default: reads from stdin).

optional arguments:

-h, --help

show this help message and exit

-a, --async

Load async pipe.

-t, --test

Run in test mode (uses default inputs).

Now to execute flow.py, type the command run-pipe flow. You should then see the following output in your terminal:

{'content': 'https://google.com', 'strreplace': 'https://google.co.uk'}

run-pipe will also search the examples directory for workflows. Type run-pipe demo and you should see the following output:

Deadline to clear up health law eligibility near
682