python-asyncio TypeError：对象字典不能用于“等待”表达式

Posted 2023-03-30

【中文标题】python-asyncio TypeError：对象字典不能用于“等待”表达式

【英文标题】：python-asyncio TypeError: object dict can't be used in 'await' expression

【发布时间】：2018-04-13 17:45:16

【问题描述】：

我正在使用第三方模块从 API 检索数据。我只是想异步等待模块返回偶尔需要几秒钟并冻结我的应用程序的数据。但是，当我尝试等待对该模块的调用时，我收到 TypeError：

TypeError: object dict can't be used in 'await' expression

import thirdPartyAPIwrapper

async def getData():
    retrienveData = await thirdPartyAPIWrapper.data()
    return await retrieveData

def main():
    loop = asncio.get_event_loop()
    data = loop.run_until_complete(getData())
    loop.close
    return data

为什么我不能等待一个类型（'dict'）？有没有解决的办法？ 如果带有 asyncio 的 async/await 不能与不返回协程的第三方模块一起使用，那么我的其他选择是什么？

【问题讨论】：

是的，在查看该模块的源代码时，它是一个普通的def 函数

【参考方案1】：

只能等待异步（使用async def 定义）函数。整个想法是这样的函数是用特殊的方式编写的，这样可以在不阻塞事件循环的情况下运行（await）它们。

如果您想从公共（使用def 定义）函数中获取结果，该函数需要相当长的时间才能执行，您可以使用以下选项：

将整个函数重写为异步 在另一个线程中调用此函数并异步等待结果 在另一个进程中调用此函数并异步等待结果

通常你想选择第二个选项。

以下是如何做到这一点的示例：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor


_executor = ThreadPoolExecutor(1)


def sync_blocking():
    time.sleep(2)


async def hello_world():
    # run blocking function in another thread,
    # and wait for it's result:
    await loop.run_in_executor(_executor, sync_blocking)


loop = asyncio.get_event_loop()
loop.run_until_complete(hello_world())
loop.close()

请阅读 this answer 了解 asyncio 的工作原理。我想它会对你有很大帮助。

【讨论】：

请注意，与 javascript 不同，Python 在创建时并未考虑到异步性。在 Python 中，如果你做一些异步的事情，你需要显式地做：特别是显式地定义异步函数。普通函数不能不努力就变成异步的。这就是为什么您需要线程来异步运行同步函数的原因。

【参考方案2】：

由于thirdPartyAPIWrapper.data() 是一个普通的同步函数，你应该在另一个线程中调用它。

asgiref 库中有一个辅助函数。 假设我们有一个带有参数的阻塞函数：

import asyncio
import time

from asgiref.sync import sync_to_async


def blocking_function(seconds: int) -> str:
    time.sleep(seconds)
    return f"Finished in seconds seconds"

async def main():
    seconds_to_sleep = 5
    function_message = await sync_to_async(blocking_function)(seconds_to_sleep)
    print(function_message)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

该库中还有一个 async_to_sync 辅助函数。

【参考方案3】：

你不需要等待哪个函数在异步运行

async def getData():
    retrienveData = thirdPartyAPIWrapper.data()
    return retrieveData

【参考方案4】：

我正在编写测试用例，我需要模拟异步功能。所以，你可以像这样写一个简单的辅助函数。

async def resolve(val):
    return val

现在你可以等待任何事情

foo = resolve(1) 
await foo # No error!

【讨论】：

不解决来自第三方的长时间运行的同步过程