aioflows

异步actor框架

文档： https://aioflows.github.io

源代码： https://github.com/apatrushev/aioflows

本项目旨在创建一个支持库，用于在Python中使用结构化数据流和actor概念构建异步应用程序。当前阶段纯粹是概念验证，作为与同事和社区讨论的基础。它不适用于任何生产或个人项目。

最小化工作示例

import asyncio

from aioflows.simple import Printer, Ticker


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


asyncio.run(start())

UDP回声示例

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())

您可以使用socat进行测试

socat - UDP:localhost:5353

其他示例

更多示例可以在 src/examples 中找到。

安装

• 本地

pip install .

• 可编辑

pip install -e .

• 开发

pip install -e .[dev]

• 示例依赖项

pip install -e .[examples]

• 全部

pip install -e .[dev,examples]

• 来自GitHub

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

常规开发步骤

运行检查和测试

inv isort flake test

运行示例（目前所有ERRORCODE的值应为0/OK或超时）

inv examples | grep ERRORCODE

类似项目

我发现了一些几乎与这个概念相同的现有解决方案

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