异步

用于Event驱动的应用程序的异步python

安装

pip install asynced

用法

asynced谎言的核心StateVar。它可以（但不是必须）包含一个值，可以监视更改，并且可以轻松地映射到其他 StateVar实例。

为简洁起见，假设下一个示例从异步函数中运行，因为StateVar需要一个正在运行的事件循环。.

创建一个空状态变量很简单：

>>> from asynced import StateVar
>>> spam = StateVar()

我们可以检查它是否设置：

>>> spam.is_set
False

让我们创建一个asyncio.Task在设置后打印垃圾邮件值。

>>> async def print_spam():
...     print(await spam)
... 
... task = asyncio.create_task(print_spam)

在我们设置 statevar 之前，什么都不会打印：

>>> spam.set('Spam!')
Spam!

在我们设置 statevar 之后，我们看到它print_spam打印了值。

到目前为止，我们没有做任何asyncio.Future不能做的事情。但是 an asyncio.Future只能设置一次，而 aStateVar可以根据需要设置为不同的值。

我们可以通过异步迭代来观察 statevar 的任何变化：

>>> async def print_spam_changes():
...     async for value in spam:
...         print(value)
... 
... task = asyncio.create_task(print_spam_changes)

现在任何时候我们设置spam一个不同的值，它都会被打印出来：

>>> spam.set('Spam and ham!')
Spam and ham!
>>> spam.set('Spam and eggs!')
Spam and eggs!

整齐吧？

AStateVar也可以通过传递异步迭代来构造，使其行为更像一个asyncio.Task：

>>> async def slowrange(*args):
...     for i in range(*args):
...         await asyncio.sleep(1)
...         yield i
...
>>> count4 = StateVar(slowrange(4))
>>> await asyncio.sleep(3.14)
>>> await count4
2

正如我们在这里看到的，statevar 将在后台自动将自己设置为来自异步迭代的值。

映射

StateVar也可以通过将函数应用于另一个来构造 StateVar：

>>> count4_inv = StateVar(slowrange(4)).map(lambda i: -i)
>>> async for i in count4_inv:
...     print(i)

现在以一秒的间隔打印0, -1,-2和。-3

StateVar.map仅适用于具有单个参数的函数，即一次用于一个状态变量。但不要担心，可以通过使用将多个状态变量映射在一起asynced.statefunction。它将任何函数 (a: A, b: B, ...) -> R转换为接受State's（State是 的超类StateVar）并返回一个新的StateVar（或其他一些子类State）的函数， (a: State[A], b: State[B], ...) -> StateVar[R]。

>>> from asynced import statefunction
>>> @statefunction
... def sadd(_a: float, _b: float) -> float:
...    return _a + _b
... 
>>> a, b = StateVar(), StateVar()
>>> c = sadd(a, b)

在这里，c将设置当且仅当且仅当a且b已设置：

>>>import calendar c.is_set
False
>>> a.set(12)
c.is_set
False
>>> b.set(30)
>>> await c
42

现在，每次那个a或b改变，c也会改变。

StateTuple

与 相同的方式StateVar，StateTuple可以等待 a 获取当前值一旦可用，并异步迭代以获取设置后的值。此外，它还可以用作单个StateVar 实例的元组。

>>> st = StateTuple(2)  # equivalent StateTuple(StateVar(), StateVar())
>>> st[0].set('spam')
>>> st[1].set('ham')
>>> await st
('spam', 'ham')
>>> await st[-1]
'ham'
>>> st[1] = 'eggs'  # equivalent to st[1].set('eggs')
>>> await st
('spam', 'eggs')
>>> s0, s1 = st
>>> await s1
'eggs'
>>> st2 = 2 * st  # equivalent to StateTuple((st[0], st[1], st[0], st[1]))
>>> st2[2] = 'bacon'
>>> await st2
('bacon', 'eggs', 'bacon', 'eggs')
>>> await st
('bacon', 'eggs')

StateDict

就像StateTuple， aStateDict是个人的集合StateVar。它更类似于 acollections.defaultdict而不是常规的 dict，因为访问没有值的键将返回一个空的StateVar。

>>> sd = StateDict()
>>> await sd
{}
>>> sd['spam'] = 'Spam!'
>>> await sd
{'spam': 'Spam!'}
>>> ham = sd['ham']
>>> ham.is_set
False
>>> list(sd.keys())
['spam']
>>> sd['ham'] = 'Ham and eggs!'
>>> await ham
'Ham and eggs!'

.get()

StateVar，StateTuple并且StateDict都实现了一个.get()方法，可以用来获取当前值，而不必使用await. 如果没有设置任何值，它将引发一个LookupError. 或者，您可以传递一个default值，该值将在未设置值的情况下返回。

错误处理

如果用于创建例如 a 的异步迭代StateVar引发错误，则.is_error属性将变为True，并且在等待它时.get或.map在调用或时引发错误。异常将传播到其映射的“孩子” State。

如果异步迭代被耗尽（或StopAsyncIteration直接引发）， .is_stopped将被设置为True，并且StopAsyncIteration只会为and的服务员重新引发。__aiter____anext__

API 参考

*~即将推出~*