aiostream

基于生成器的异步迭代操作符

概述

aiostream_提供了一系列流操作符，可以组合创建异步操作管道。

它可以看作是itertools_的异步版本，尽管某些方面略有不同。本质上，所有提供的操作符都返回一个统一的接口，称为流。流是一个增强的异步可迭代对象，提供以下功能:

操作符管道 - 使用管道符号 |
可重复性 - 每次迭代创建一个不同的迭代器
安全的迭代上下文 - 使用 async with 和 stream 方法
简化的执行 - 使用 await 获取流的最后一个元素
切片和索引 - 使用方括号 []
连接 - 使用加号 +

流操作符

流操作符_分为7类:

+--------------------+---------------------------------------------------------------------------------------+ | 创建 | iterate_, preserve_, just_, call_, empty_, throw_, never_, repeat_, count_, range_ | +--------------------+---------------------------------------------------------------------------------------+ | 转换 | map_, enumerate_, starmap_, cycle_, chunks_ | +--------------------+---------------------------------------------------------------------------------------+ | 选择 | take_, takelast_, skip_, skiplast_, getitem_, filter_, until_, takewhile_, dropwhile_ | +--------------------+---------------------------------------------------------------------------------------+ | 组合 | map_, zip_, merge_, chain_, ziplatest_ | +--------------------+---------------------------------------------------------------------------------------+ | 聚合 | accumulate_, reduce_, list_ | +--------------------+---------------------------------------------------------------------------------------+ | 高级 | concat_, flatten_, switch_, concatmap_, flatmap_, switchmap_ | +--------------------+---------------------------------------------------------------------------------------+ | 时间 | spaceout_, timeout_, delay_ | +--------------------+---------------------------------------------------------------------------------------+ | 其他 | action_, print_ | +--------------------+---------------------------------------------------------------------------------------+

示例

以下示例展示了流的大部分功能:

.. code:: python

import asyncio
from aiostream import stream, pipe


async def main():

    # 创建一个间隔0.2秒的计数流
    xs = stream.count(interval=0.2)

    # 操作符可以使用'|'管道
    ys = xs | pipe.map(lambda x: x**2)

    # 流可以切片
    zs = ys[1:10:2]

    # 使用流上下文进行适当的资源管理
    async with zs.stream() as streamer:

        # 异步迭代
        async for z in streamer:

            # 打印 1, 9, 25, 49 和 81
            print('->', z)

    # 流可以被等待并返回最后一个值
    print('9² = ', await zs)

    # 流可以多次运行
    print('9² = ', await zs)

    # 流可以连接
    one_two_three = stream.just(1) + stream.range(2, 4)

    # 打印 [1, 2, 3]
    print(await stream.list(one_two_three))


# 运行主协程
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

更多示例可以在文档的示例部分_中找到。