A functional and reactive framework for RxPY.
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:
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 reactivex as rx import reactivex.operators as ops EchoSource = namedtuple('EchoSource', ['httpd']) EchoSink = namedtuple('EchoSink', ['httpd']) EchoDrivers = namedtuple('EchoDrivers', ['httpd']) def echo_server(source): init = rx.from_([ httpd.Initialize(), httpd.AddRoute(methods=['GET'], path='/echo/{what}', id='echo'), httpd.StartServer(host='localhost', port=8080), ]) echo = source.httpd.route.pipe( ops.filter(lambda i: i.id == 'echo'), ops.flat_map(lambda i: i.request), ops.map(lambda i: httpd.Response( context=i.context, data=i.match_info['what'].encode('utf-8')), ) ) control = rx.merge(init, echo) return EchoSink(httpd=httpd.Sink(control=control)) def main(): run(Component(call=echo_server, input=EchoSource), EchoDrivers(httpd=httpd.make_driver())) if __name__ == '__main__': 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 hello
Cyclotron is available on PyPi and can be installed with pip:
pip install cyclotronCyclotron automatically uses uvloop if it is available.
This project is composed of several python packages. Install also the ones that you use in your application:
| Package | Version |
|---|---|
| cyclotron | |
| cyclotron-std |  |
| cyclotron-aiohttp |  |
| cyclotron-aiokafka |  |
| cyclotron-consul |  |
This project is licensed under the MIT License - see the License file for details