Skip to main content

A reactive stream cycle implementation in python

Project description

A functional and reactive framework for RxPY. Documentation Status

With Cyclotron, you can structure your RxPY code as many reusable components. Moreover it naturally encourages to separate pure code and side effects. So a Cyclotron application is easier to test, maintain, and extend.

Here is the structure of a cyclotron application:

How to use it

The following example is an http echo server:

from collections import namedtuple

from cyclotron import Component
from cyclotron.asyncio.runner import run
import cyclotron_aiohttp.httpd as httpd
import rx
import rx.operators as ops

EchoSource = namedtuple('EchoSource', ['httpd'])
EchoSink = namedtuple('EchoSink', ['httpd'])
EchoDrivers = namedtuple('EchoDrivers', ['httpd'])

def echo_server(source):
    init = rx.from_([
        httpd.AddRoute(methods=['GET'], path='/echo/{what}', id='echo'),
        httpd.StartServer(host='localhost', port=8080),

    echo = source.httpd.route.pipe(
        ops.filter(lambda i: == 'echo'),
        ops.flat_map(lambda i: i.request), i: httpd.Response(

    control = rx.merge(init, echo)
    return EchoSink(httpd=httpd.Sink(control=control))

def main():
    run(Component(call=echo_server, input=EchoSource),

if __name__ == '__main__':

In this application, the echo_server function is a pure function, while the http server is implemented as a driver.

pip install cyclotron-aiohttp

you can then test it with an http client like curl:

$ curl http://localhost:8080/echo/hello


Cyclotron is available on PyPi and can be installed with pip:

pip install cyclotron

This project is composed of several python packages. Install also the ones that you use in your application:

Package Version
cyclotron pypi-cyclotron
cyclotron-std pypi-cyclotron-std
cyclotron-aiohttp pypi-cyclotron-aiohttp


This project is licensed under the MIT License - see the License file for details

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for cyclotron, version 1.2.1
Filename, size File type Python version Upload date Hashes
Filename, size cyclotron-1.2.1.tar.gz (8.3 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring DigiCert DigiCert EV certificate Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page