AsyncED

异步 Python 用于 事件 - 驱动 应用

安装

pip install asynced

用法

在 asynced 的核心是 StateVar。它可以（但不一定）包含一个值，可以观察其变化，并且可以轻松地映射到其他 StateVar 实例。

为了简洁起见，以下示例假定是在异步函数中运行的，因为 StateVar 需要运行的事件循环。.

创建一个空的状态变量就像这样

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

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

>>> spam.is_set
False

让我们创建一个 asyncio.Task，当状态变量设置时，它将打印 spam 值。

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

在设置状态变量之前，没有打印任何内容

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

在设置状态变量后，我们看到 print_spam 打印了值。

到目前为止，我们做的任何事情 asyncio.Future 都可以做到。但是，asyncio.Future 只能设置一次，而 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!

好看吗？

StateVar 也可以通过传递异步可迭代对象来构建，使其表现得更像一个 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 来构建

>>> 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 的（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)

在这里，只有当 a 和 b 都设置时，c 才会被设置

>>>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 可以被等待以获取当前值，一旦可用，可以通过异步迭代来获取已设置的值。此外，它还可以用作单个 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 类似，一个 StateDict 是单个 StateVar 的集合。它更类似于 collections.defaultdict 而不是常规字典，因为访问未设置值的键将返回一个空的 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 值，当没有设置值时返回。

错误处理

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

如果异步可迭代对象耗尽（或直接引发 StopAsyncIteration），则将 .is_stopped 设置为 True，并且 StopAsyncIteration 将仅对 __aiter__ 和 __anext__ 的等待者重新引发。

API 参考

*~ 即将推出 ~*