如何在 python lambda 中使用 await

Posted 2023-04-18

【中文标题】如何在 python lambda 中使用 await

【英文标题】：How to use await in a python lambda

【发布时间】：2017-04-06 09:14:54

【问题描述】：

我正在尝试做这样的事情：

mylist.sort(key=lambda x: await somefunction(x))

但我收到此错误：

SyntaxError: 'await' outside async function

这是有道理的，因为 lambda 不是异步的。

我尝试使用async lambda x: ...，但会抛出SyntaxError: invalid syntax。

Pep 492 状态：

可以提供异步 lambda 函数的语法，但此构造超出了本 PEP 的范围。

但我无法确定该语法是否在 CPython 中实现。

有没有办法声明一个异步 lambda，或者使用一个异步函数对列表进行排序？

【参考方案1】：

你不能。没有async lambda，即使有，您也不能将它作为键函数传递给list.sort()，因为键函数将作为同步函数调用而不是等待。一个简单的解决方法是自己注释您的列表：

mylist_annotated = [(await some_function(x), x) for x in mylist]
mylist_annotated.sort()
mylist = [x for key, x in mylist_annotated]

请注意，列表推导中的 await 表达式仅在 Python 3.6+ 中受支持。如果您使用的是 3.5，则可以执行以下操作：

mylist_annotated = []
for x in mylist:
    mylist_annotated.append((await some_function(x), x)) 
mylist_annotated.sort()
mylist = [x for key, x in mylist_annotated]

【讨论】：

我得到了一个SyntaxError: 'await' expressions in comprehensions are not supported，所以我必须这样做（以供将来参考）：mylist_annotated = [] for x in mylist: mylist_annotated.append((await some_function(x), x) ) mylist_annotated.sort() mylist = [x for key, x in mylist_annotated] 现在它可以工作了，谢谢！

@iCart 对，这是 Python 3.5 中的限制，在即将推出的 Python 3.6 中取消了。

发现了一个非常边缘的案例 - 请参阅我的答案 :-)

【参考方案2】：

可以通过将lambda 与async 生成器组合来模拟“asynclambda”：

key=lambda x: (await somefunction(x) for _ in '_').__anext__()

可以将( ).__anext__() 移动到帮助器，这也可能使模式更清晰：

def head(async_iterator): return async_iterator.__anext__()

key=lambda x: head(await somefunction(x) for _ in '_')

请注意，标准库中的排序方法/函数不是异步的。需要一个异步版本，例如asyncstdlib.sorted（免责声明：我维护这个库）：

import asyncstdlib as a

mylist = await a.sorted(mylist, key=lambda x: head(await somefunction(x) for _ in '_'))

---

了解lambda ...: (...).__anext__() 模式

“asynclambda”将是一个匿名异步函数，或者换句话说，一个匿名函数评估为等待。这与 async def 如何定义一个命名函数评估为可等待对象是平行的。 该任务可以分为两部分：一个匿名函数表达式和一个嵌套等待表达式。

匿名函数表达式正是lambda ...: ...。

一个等待的表达式is only allowed inside a coroutine function；但是：

（异步）生成器表达式隐式创建（协程）函数。由于异步生成器只需要异步运行，因此可以在同步函数 (since Python 3.7) 中定义。 可以通过其__anext__ method 将异步可迭代对象用作可等待对象。

这三个部分直接用在“asynclambda”模式中：

#   | regular lambda for the callable and scope
#   |         | async generator expression for an async scope
#   v         v                                    v first item as an awaitable
key=lambda x: (await somefunction(x) for _ in '_').__anext__()

异步生成器中的for _ in '_' 只能进行一次迭代。任何至少有一次迭代的变体都可以。

【参考方案3】：

await 不能包含在 lambda 函数中。

这里的解决方案可以简化为：

from asyncio import coroutine, run


my_list = [. . .]


async def some_function(x) -> coroutine:
    . . .

my_list.sort(key=lambda x: await some_function(x))  # raises a SyntaxError
my_list.sort(key=lambda x: run(some_function(x))  # works

【参考方案4】：

如果你已经定义了一个单独的异步函数，你可以进一步简化 MisterMiyagi 的回答：

mylist = await a.sorted(
    mylist, 
    key=somefunction)

如果你想在等待它之后改变key，你可以使用asyncstdlib.apply：

mylist = await a.sorted(
    mylist, 
    key=lambda x: a.apply(lambda after: 1 / after, some_function(x)))

这是一个完整的示例程序：

import asyncio
import asyncstdlib as a

async def some_function(x):
    return x

async def testme():
    mylist=[2, 1, 3]

    mylist = await a.sorted(
        mylist, 
        key=lambda x: a.apply(lambda after: 1 / after, some_function(x)))
        
    print(f'mylist is: mylist')
    

if __name__ == "__main__":
    asyncio.run(testme())

【参考方案5】：

Sven Marnach 的回答有一个边缘案例。

如果您尝试对包含 2 个生成相同搜索键但不同且不可直接排序的项目的列表进行排序，它将崩溃。

mylist = ['score':50,'name':'bob','score':50,'name':'linda']

mylist_annotated = [(x['score'], x) for x in mylist]
mylist_annotated.sort()
print( [x for key, x in mylist_annotated] )

将给予：

TypeError: '<' not supported between instances of 'dict' and 'dict'

幸运的是，我有一个简单的解决方案 - 我的数据有一个可排序的唯一键，因此我可以将其作为第二个键：

mylist = ['score':50,'name':'bob','unique_id':1,'score':50,'name':'linda','unique_id':2]

mylist_annotated = [(x['score'], x['unique_id'], x) for x in mylist]
mylist_annotated.sort()
print( [x for key, unique, x in mylist_annotated] )

我猜如果您的数据没有自然唯一的值，您可以在尝试排序之前插入一个？也许是一个 uuid？

编辑：正如评论中的建议（谢谢！），您也可以使用 operator.itemgetter：

import operator

mylist = ['score':50,'name':'bob','score':50,'name':'linda']

mylist_annotated = [(x['score'], x) for x in mylist]
mylist_annotated.sort(key=operator.itemgetter(0))
print( [x for key, x in mylist_annotated] )

【讨论】：

我认为这种边缘情况的最佳解决方案是将operator.itemgetter(0) 作为关键函数传递给sort()。元组按字典顺序排序，因此相同的键将导致第二项的比较。通过仅显式选择第一项作为排序键，我们可以防止第二次比较。