asyncio嵌套字典推导使用await报错,求更优雅实现方案
错误原因
外层是同步字典推导,内层试图在普通字典推导中使用await。Python的异步推导(如async {k: v ...})必须显式声明async,且只能搭配异步迭代器使用,但user_rooms.items()是同步迭代器,无法用async for遍历,所以直接嵌套异步推导的写法不成立。
更优雅的实现方式
方式1:用字典推导替代zip,提升可读性
原方案用zip+dict组装内层字典,换成普通字典推导会更直观,键值对应关系一目了然:
async def func(): res = {} for shard, shard_rooms in user_rooms.items(): # 并行获取当前shard下所有room的结果 room_results = await asyncio.gather( *[self.room_objects(room=room, shard=shard) for room in shard_rooms] ) # 用字典推导组装内层字典 res[shard] = {room: result for room, result in zip(shard_rooms, room_results)} return res
方式2:并行处理所有shard,提升效率
如果不同shard的请求之间没有依赖,可以把每个shard的处理封装成协程,用asyncio.gather并行执行所有shard的任务,比原方案串行处理shard的效率更高:
async def fetch_shard_data(shard, rooms): room_results = await asyncio.gather( *[self.room_objects(room=room, shard=shard) for room in rooms] ) return shard, {room: result for room, result in zip(rooms, room_results)} async def func(): # 为每个shard创建处理任务 tasks = [fetch_shard_data(shard, rooms) for shard, rooms in user_rooms.items()] # 并行执行所有任务,结果直接转成字典 res = dict(await asyncio.gather(*tasks)) return res
方式3:Python 3.10+ 异步推导的合理用法(可选)
如果你的Python版本是3.10+,可以用异步字典推导,但需要注意:异步推导本身是协程,且要保证结果和room的顺序对应,所以还是要结合gather来并行执行,写法如下:
async def func(): async def build_shard_dict(shard, rooms): results = await asyncio.gather(*[self.room_objects(room=r, shard=shard) for r in rooms]) return shard, {r: res for r, res in zip(rooms, results)} tasks = [build_shard_dict(shard, rooms) for shard, rooms in user_rooms.items()] res = dict(await asyncio.gather(*tasks)) return res
这种写法和方式2本质类似,只是把内层逻辑封装到了异步函数里。
总结
原方案的核心问题是内层字典组装的可读性不足,优化方向主要有两个:
- 用普通字典推导替代
zip+dict,让键值对应关系更清晰; - 若shard之间无依赖,用
asyncio.gather并行处理所有shard的任务,提升整体执行效率。
内容的提问来源于stack exchange,提问作者Mofeng
相关产品推荐
相关产品推荐

