riko 0.29.0

Notes

Optional Dependencies

Notes

Why I built riko

Yahoo! Pipes [4] was a user friendly web application used to “aggregate, manipulate, and mashup content from around the web.” Wanting to create custom pipes, I came across pipe2py which translated a Yahoo! Pipes into python code. pipe2py suited my needs at the time but was unmaintained and lacked asynchronous and parallel processing APIs.

riko addresses the shortcomings of pipe2py but breaks compatibility with Yahoo! Pipes workflows. riko contains ~40 built-in modules, aka pipes, that allow you to programatically recreate much of what you previously could on Yahoo! Pipes.

Why you should use riko

riko provides a number of benefits / differences from other stream processing applications such as Huginn, Flink, Spark, and Storm [5]. Namely:

• a small footprint (CPU and memory usage)

• native RSS support

• simple installation and usage

• pypy support

• modular pipes to filter, sort, and modify feeds

The subsequent tradeoffs riko makes are:

• not distributed (able to run on a cluster of servers)

• no GUI for creating workflows

• doesn’t continually monitor feeds for new data

• can’t react to specific events

• iterator (pull) based so only supports a single consumer

The following table summaries these observations:

For more detailed information, please check-out the FAQ.

Notes

Usage Index

• Fetching feeds

• Synchronous processing

• Parallel processing

• Asynchronous processing

• Cookbook

Fetching feeds

riko can read both local and remote filepaths via source pipes. All source pipes return an equivalent feed iterator of dictionaries, aka items.

>>> from itertools import chain
>>> from riko import get_path
>>> from riko.modules.pipefetch import pipe as fetch
>>> from riko.modules.pipefetchpage import pipe as fetchpage
>>> from riko.modules.pipefetchdata import pipe as fetchdata
>>> from riko.modules.pipefetchsitefeed import pipe as fetchsitefeed
>>> from riko.modules.pipefeedautodiscovery import pipe as autodiscovery
>>>
>>> ### Fetch a url ###
>>> feed = fetchpage(conf={'url': 'https://news.ycombinator.com'})
>>>
>>> ### Fetch a filepath ###
>>> #
>>> # Note: `get_path` just looks up a file in the `data` directory
>>> feed = fetchdata(conf={'url': get_path('quote.json')})
>>>
>>> ### View the fetched data ###
>>> item = next(feed)
>>> item['list']['resources'][0]['resource']['fields']['symbol']
'KRW=X'

>>> ### Fetch an rss feed ###
>>> feed = fetch(conf={'url': 'https://news.ycombinator.com/rss'})
>>>
>>> ### Fetch the first rss feed found ###
>>> feed = fetchsitefeed(conf={'url': http://www.bbc.com/news})
>>>
>>> ### Find all rss links and fetch the feeds ###
>>> entries = autodiscovery(conf={'url': http://edition.cnn.com/services/rss/})
>>> urls = (e['link'] for e in entries)
>>> feed = chain.from_iterable(fetch(conf={'url': url}) for url in urls)
>>>
>>> ### Alternatively, create a SyncCollection ###
>>> #
>>> # `SyncCollection` is a url fetching convenience class with support for
>>> # parallel processing
>>> from riko.lib.collections import SyncCollection
>>>
>>> sources = [{'url': url} for url in urls]
>>> feed = SyncCollection(sources).fetch()
>>>
>>> ### View the fetched rss feed(s) ###
>>> #
>>> # Note: regardless of how you fetch an rss feed, it will have the same
>>> # structure
>>> item = next(feed)
>>> sorted(item.keys())
[
    'author', 'author.name', 'author.uri', 'comments', 'content',
    'dc:creator', 'id', 'link', 'pubDate', 'summary', 'title',
    'updated', 'updated_parsed', 'y:id', 'y:published', 'y:title']
>>> item['title'], item['author'], item['link']
(
    u'Using NFC tags in the car', u'Liam Green-Hughes',
    u'http://www.greenhughes.com/content/using-nfc-tags-car')

Please see the FAQ for a complete list of supported file types and protocols

Synchronous processing

riko can modify feeds by combining any of the 40 built-in pipes

>>> from itertools import chain
>>> from riko import get_path
>>> from riko.modules.pipefetch import pipe as fetch
>>> from riko.modules.pipefilter import pipe as pfilter
>>> from riko.modules.pipesubelement import pipe as subelement
>>> from riko.modules.piperegex import pipe as regex
>>> from riko.modules.pipesort import pipe as sort
>>>
>>> ### Set the pipe configurations ###
>>> #
>>> # Notes:
>>> #   1. `get_path` just looks up a file in the `data` directory
>>> #   2. the `dotall` option is used to match `.*` across newlines
>>> fetch_conf = {'url': get_path('feed.xml')}                                          # 1
>>> filter_rule = {'field': 'y:published', 'op': 'before', 'value': '2/5/09'}
>>> sub_conf = {'path': 'content.value'}
>>> match = r'(.*href=")([\w:/.@]+)(".*)'
>>> regex_rule = {'field': 'content', 'match': match, 'replace': '$2', 'dotall': True}  # 2
>>> sort_conf = {'rule': {'sort_key': 'content', 'sort_dir': 'desc'}}
>>>
>>> ### Create a SyncPipe workflow ###
>>> #
>>> # `SyncPipe` is a workflow convenience class that enables method
>>> # chaining and parallel processing.
>>> #
>>> # The following workflow will:
>>> #   1. fetch the rss feed
>>> #   2. filter for items published before 2/5/2009
>>> #   3. extract the path `content.value` from each feed item
>>> #   4. replace the extracted text with the last href url contained
>>> #      within it
>>> #   5. reverse sort the items by the replaced url
>>> #   5. return the raw feed iterator
>>> #
>>> # Note: sorting is not lazy so take caution when using this pipe
>>> from riko.lib.collections import SyncPipe
>>>
>>> output = (SyncPipe('fetch', conf=fetch_conf)  # 1
...     .filter(conf={'rule': filter_rule})       # 2
...     .subelement(conf=sub_conf, emit=True)     # 3
...     .regex(conf={'rule': regex_rule})         # 4
...     .sort(conf=sort_conf)                     # 5
...     .output)                                  # 6
>>>
>>> next(output)
{'content': 'mailto:mail@writetoreply.org'}

Please see Design Principles for an alternative (function based) workflow. Please see pipes for a complete list of available pipes.

Parallel processing

An example using riko’s ThreadPool based parallel API

>>> from riko import get_path
>>> from riko.lib.collections import SyncPipe
>>>
>>> ### Set the pipe configurations ###
>>> #
>>> # Notes:
>>> #   1. `get_path` just looks up a file in the `data` directory
>>> #   2. the `dotall` option is used to match `.*` across newlines
>>> url = get_path('feed.xml')                                                          # 1
>>> filter_rule1 = {'field': 'y:published', 'op': 'before', 'value': '2/5/09'}
>>> match = r'(.*href=")([\w:/.@]+)(".*)'
>>> regex_rule = {'field': 'content', 'match': match, 'replace': '$2', 'dotall': True}  # 2
>>> filter_rule2 = {'field': 'content', 'op': 'contains', 'value': 'file'}
>>> strtransform_conf = {'rule': {'transform': 'rstrip', 'args': '/'}}
>>>
>>> ### Create a parallel SyncPipe workflow ###
>>> #
>>> # The following workflow will:
>>> #   1. fetch the rss feed
>>> #   2. filter for items published before 2/5/2009
>>> #   3. extract the path `content.value` from each feed item
>>> #   4. replace the extracted text with the last href url contained
>>> #      within it
>>> #   5. filter for items with local file urls (which happen to be rss
>>> #      feeds)
>>> #   6. strip any trailing `\` from the url
>>> #   7. remove duplicate urls
>>> #   8. fetch each rss feed
>>> #   9. Merge the rss feeds into a list
>>> feed = (SyncPipe('fetch', conf={'url': url}, parallel=True)  # 1
...     .filter(conf={'rule': filter_rule1})                     # 2
...     .subelement(conf=sub_conf, emit=True)                    # 3
...     .regex(conf={'rule': regex_rule})                        # 4
...     .filter(conf={'rule': filter_rule2})                     # 5
...     .strtransform(conf=strtransform_conf)                    # 6
...     .uniq(conf={'uniq_key': 'strtransform'})                 # 7
...     .fetch(conf={'url': {'subkey': 'strtransform'}})         # 8
...     .list)                                                   # 9
>>>
>>> len(feed)
25

Asynchronous processing

To enable this asynchronous processing, you must install the async module.

pip install riko[async]

An example using riko’s optional Twisted powered asynchronous API.

>>> from twisted.internet.task import react
>>> from twisted.internet.defer import inlineCallbacks
>>> from riko import get_path
>>> from riko.twisted.collections import AsyncPipe
>>>
>>> ### Set the pipe configurations ###
>>> #
>>> # Notes:
>>> #   1. `get_path` just looks up a file in the `data` directory
>>> #   2. the `dotall` option is used to match `.*` across newlines
>>> url = get_path('feed.xml')                                                          # 1
>>> filter_rule1 = {'field': 'y:published', 'op': 'before', 'value': '2/5/09'}
>>> match = r'(.*href=")([\w:/.@]+)(".*)'
>>> regex_rule = {'field': 'content', 'match': match, 'replace': '$2', 'dotall': True}  # 2
>>> filter_rule2 = {'field': 'content', 'op': 'contains', 'value': 'file'}
>>> strtransform_conf = {'rule': {'transform': 'rstrip', 'args': '/'}}
>>>
>>> ### Create a AsyncPipe workflow ###
>>> #
>>> # See `Parallel processing` above for an explanation of the steps this
>>> # performs
>>> @inlineCallbacks
... def run(reactor):
...     feed = yield (AsyncPipe('fetch', conf={'url': url})
...         .filter(conf={'rule': filter_rule1})
...         .subelement(conf=sub_conf, emit=True)
...         .regex(conf={'rule': regex_rule})
...         .filter(conf={'rule': filter_rule2})
...         .strtransform(conf=strtransform_conf)
...         .uniq(conf={'uniq_key': 'strtransform'})
...         .fetch(conf={'url': {'subkey': 'strtransform'}})
...         .list)
...
...     print(len(feed))
...
>>> react(run)
25

Cookbook

Please see the cookbook or ipython notebook for more examples.

Notes

Setup

pip install -r dev-requirements.txt

Examples

Run python linter and nose tests

manage lint
manage test