asyncio_utils 0.1.4

功能

• 异步实用程序

安装

要安装：

pip install asyncio-utils

用法

几乎所有的内容都在前面与await关键字一起使用，除非标记了 否则。但是，大多数方法输入可以是awaitable（但不是 实际上还等着呢）而且它们仍然可以工作，除非另有标记。

运行任何示例：

import asyncio
loop = asyncio.get_event_loop()

艾特

包装/确保AsyncIterator。

如果输入是Awaitable，那么我们将await结果，并检查 如果它返回并且AsyncIterator。

如果输入是未调用的async generator，则我们将 调用它并给出它的值。

否则，如果输入是iterator，我们将对其进行迭代并生成值。

示例：

>>> async def main():
    async for v in aiter2(range(1, 5)):  # normal iterator
        print(v)

>>> loop.run_until_complete(main())
1
2
3
4

>>> async def main():
        async for v in aiter2(arange(1, 5)):  # not awaited
            print(v)

>>> loop.run_until_complete(main())
1
2
3
4

>>> async def main():
        async for v in aiter2(await arange(1, 5)):  # awaited works
            print(v)

>>> loop.run_until_complete(main())
1
2
3
4

>>> async def agen():
        yield 1
        yield 2
        yield 3
        yield 4

>>> async def main():
        async for v in aiter2(agen):  # oops forgot to call it
            print(v)

>>> loop.run_until_complete(main())
1
2
3
4

下一步

模拟内置的next方法。此方法不接受 awaitable。输入必须是AsyncIterator，否则将得到 TypeError。

示例：

>>> async def main():
    myrange = await arange(1, 5)
    for n in range(1, 5):
        print(n, n == await anext(myrange))
    try:
        n = await anext(myrange)
        print("This should not be shown")
    except StopAsyncIteration:
        print('Sorry no more values!')

>>> loop.run_until_complete(main())
1 True
2 True
3 True
4 True
Sorry no more values!

出现StopAsyncIteration时使用默认值的示例：

>>> async def main():
    myrange = await arange(1)
    print(await anext(myrange))
    print(await anext(myrange, 'Sorry no more values!'))
    # or as kwarg
    print(await anext(myrange, default='Still no more values!'))

>>> loop.run_until_complete(main())
1
Sorry no more values!
Still no more values!

示例失败，因为传入了非AsyncIterator：

>>> async def main():
        val = await anext(arange(1, 5))
        print(val)  # never get here

>>> loop.run_until_complete(main())
Traceback (most recent call last):
...
TypeError: Not an AsyncIterator: <coroutine object arange at 0x1068170f8>

amap

AsyncGenerator模拟内置的map方法。

示例：

>>> async def main():
        async for val in amap('${}'.format, arange(1, 5)):
            print(val)

>>> loop.run_until_complete(main())
$1
$2
$3
$4

如果传入的函数是协同程序，则此操作也有效：

>>> async def formatter(val):
        return f'${val}'

>>> async def main():
        async for val in amap(formatter, arange(1, 5)):
            print(val)

>>> loop.run_until_complete(main())
$1
$2
$3
$4

过滤器

模拟内置方法的async generator。

示例：

>>> async def main():
        myfilter = await afilter(lambda x: x == 2, arange(1, 5))
        print(await anext(myfilter, 'Oops no more twos'))
        print(await anext(myfilter, 'Oops no more twos'))

>>> loop.run_until_complete(main())
2
Oops no more twos

范围

模拟内置的range方法。返回AsyncIterator。

示例：

>>> async def main():
        myrange = await arange(1, 5)
        async for n in myrange:
            print(n)

>>> loop.run_until_complete(main())
1
2
3
4

列表

将AsyncIterator转换为列表。这相当于：

[v async for v in async_iterator]

但是，我们确保async_iterator实际上是AsyncIterator。

示例：

>>> async def main():
        print(await alist(arange(1, 5)))
        # or
        print(await alist(await arange(1, 5)))

>>> loop.run_until_complete(main())
[1, 2, 3, 4]
[1, 2, 3, 4]

偶数

将AsyncIterator转换为tuple。这相当于：

tuple([v async for v in async_iterator])

但是，我们确保async_iterator实际上是AsyncIterator。

示例：

>>> async def main():
        print(await atuple(arange(1, 5)))
        # or
        print(await atuple(await arange(1, 5)))

>>> loop.run_until_complete(main())
(1, 2, 3, 4)
(1, 2, 3, 4)

aset

将AsyncIterator转换为set。这相当于：

{v async for v in async_iterator}

但是，我们确保async_iterator实际上是AsyncIterator。

示例：

>>> async def main():
        print(await aset(arange(1, 5)))
        # or
        print(await aset(await arange(1, 5)))

>>> loop.run_until_complete(main())
{1, 2, 3, 4}
{1, 2, 3, 4}

ADICT

将AsyncIterator转换为dict。这相当于：

{k: v async for (k, v) in async_iterator}

但是，我们确保async_iterator实际上是AsyncIterator。

示例：

>>> async def k_v_gen():
        async for n in await arange(1, 5):
            yield (n, n * 2)

>>> async def main():
        print(await adict(k_v_gen()))

>>> loop.run_until_complete(main())
{1: 2, 2: 4, 3: 6, 4: 8}

转换工厂

这可用于将AsyncIterator转换为任何可调用的。这是 alist、aset、atuple和adict的基。虽然没有测试， 理论上，您应该能够将它转换为 callable使用标准迭代器。

如何在代码中声明alist方法的示例：

>>> import functools
>>> alist = functools.partial(transform_factory, _type=list)
>>> alist.__doc__ = """Async list documentation."""

>>> async def main():
        print(await alist(arange(1, 5)))

>>> loop.run_until_complete(main())
[1, 2, 3, 4]

使异步

使任何可呼叫的等待。可以用作装饰。

示例：

>>> class AClass(object):

        def __init__(self):
            self.a = 'a'

>>> async_aclass = make(async_aclass)

# or as a decorator
>>> @make_async
    def sync_a():
        return 'a'

>>> async def main():
        async_a = await async_aclass()
        print(async_a.a == 'a')

        print(await sync_a())