如何在Python lambda中使用await

Question

我想要做类似于这样的事情：

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，或者使用异步函数对列表进行排序？

- iCart

Answer 1

你做不到。没有异步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]

- Sven Marnach

---

我遇到了一个 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版本中将取消这个限制。 - Sven Marnach

发现了一个非常特殊的情况 - 请看我的答案 :-) - James

这被称为“Schwartzian Transform” - izrik

Answer 2

通过结合lambda和async生成器，可以模拟出一个 "async lambda"：¹

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__() 模式

async lambda 将是一个匿名的异步函数，换句话说就是一个返回可等待对象的匿名函数。这与 async def 定义一个返回可等待对象的具名函数类似。
该任务可以分为两部分：一个匿名函数表达式和一个嵌套的可等待对象表达式。

• 匿名函数表达式完全就是 lambda ...: ...。

• 可等待对象表达式只能在协程函数内部使用；但是：

  • (异步) 生成器表达式隐式地创建了一个 (协程) 函数。由于异步生成器只需要 async 来运行，因此它可以在同步函数中进行定义 (自 Python 3.7 起)。
  • 通过其__anext__ 方法，可以将异步可迭代对象用作可等待对象。

这三个部分直接用于 "async lambda" 模式：

#   | 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 '_' 只是为了确保恰好进行一次迭代。任何具有至少一个迭代的变体都可以。

---

¹请注意是否实际需要“async lambda”，因为异步函数与常规函数一样是一等公民。正如 lambda x: foo(x) 是多余的，应该改为 foo，lambda x: (await bar(x) …)也是多余的，应该改为 bar 。 函数体应该做的不仅仅是调用和await，例如 3 + await bar(x) 或 await bar(x) or await qux(x).

- MisterMiyagi

---

这既美丽又有点令人不安！有没有办法使用这个语法来创建一个等效于 async def foo(): return None 的 lambda 表达式？我尝试过 lambda : (None for _ in "_").__anext__()，但这行不通。 - Frank Yellin

@FrankYellin 我担心这个技巧只有在有异步操作的情况下才有效。对于简单的“返回常量”的情况，我建议编写一个实用函数。 - MisterMiyagi

无论如何，使用实用函数可能更易读。但我还是得问一下。 - Frank Yellin

在我的情况下，我使用高阶函数进行分发或其他我无法知道哪个函数将被调用的操作。如脚注中所述，异步函数是一等公民，实际上几乎与同步函数无法区分。可以使用 lambda 制作一个“thunk”，它返回异步函数的 调用结果 ——也就是协程。 然后可以像正常一样 await 该协程。例如：await (lambda: foo_async(a, b, c))() 将完美运行。 我不认为这对 OP 的情况特别有用，但是这个 SO 是“python async lambda”的最佳结果，因此…… - pyansharp

@pyansharp，我希望脚注能够说明清楚。如果脚注或答案可以改进，请告诉我。 - MisterMiyagi

Answer 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

- Cristian-Dākuerikusā

Answer 4

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

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

如果您想在等待密钥之后更改它，可以使用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())

- marko.ristin

Answer 5

Sven Marnach的答案存在边缘情况。

如果您尝试对一个包含两个产生相同搜索键但不同且无法直接排序的项目的列表进行排序，它将会崩溃。

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] )

- James

---

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