aioflows 0.2.0

Python data flows library to build structured applications

aioflows

Asynchronous actors framework

---

Documentation: https://aioflows.github.io

Source: https://github.com/apatrushev/aioflows

---

This project aims to create a support library for constructing asynchronous applications in Python, using the concept of structured data flows and actors. The current phase is purely a proof-of-concept and serves as a basis for discussion with colleagues and the community. It is not intended for use in any production or personal projects.

Minimal working example

import asyncio

from aioflows.simple import Printer, Ticker


async def start():
    await (Ticker() >> Printer()).start()


asyncio.run(start())

Udp echo example

import asyncio

from aioflows.network import Udp
from aioflows.simple import Printer, Tee


async def start():
    udp = Udp(local_addr=('127.0.0.1', 5353), reuse_port=True)
    await (udp >> Tee(Printer()) >> udp).start()


asyncio.run(start())

You can test it with socat:

socat - UDP:localhost:5353

Other examples

More examples can be found in src/examples.

Installation

• local

pip install .

• editable

pip install -e .

• development

pip install -e .[dev]

• examples dependencies

pip install -e .[examples]

• all together

pip install -e .[dev,examples]

• from github

pip install git+https://github.com/apatrushev/aioflows.git

Usual development steps

Run checks and tests:

inv isort flake test

Run examples (all ERRORCODE's should be 0/OK or timeout at the moment):

inv examples | grep ERRORCODE

Similar projects

I found existing solutions that are almost equal to this concept:

• https://github.com/ReactiveX/RxPY
• https://github.com/vxgmichel/aiostream