A reactive stream cycle implementation in python
Project description
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:
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 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
Install
Cyclotron is available on PyPi and can be installed with pip:
pip install cyclotron
Cyclotron 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 |
---|---|
License
This project is licensed under the MIT License - see the License file for details
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for cyclotron-2.0.1-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 5dac54a26e10791ed7cd1897ad40aee40772dd4db37233497e19d90eea103579 |
|
MD5 | 86d22cc9ca6c1d50cd35e0fe0c025a3c |
|
BLAKE2b-256 | f19a749fb12f8df30f603a634380b58fe9148651cc9455c0659af115b2a7731f |