Skip to content

apatrushev/aioflows

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

66 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

aioflows

Github Actions PyPI

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: