Python 异步 : what satisfies `isinstance( (generator-based coroutune), ???) == True` ?

Question

我很困惑地发现 typing.Awaitable 都没有也不是 collections.abc.Awaitable涵盖一个基于生成器的协程，它是在

https://www.python.org/dev/peps/pep-0492/#await-expression

从 Python 3.6 开始，有几个 asyncio API如sleep()和 open_connection()实际上返回基于生成器的协程。我一般申请都没有问题await返回值的关键字，但我将处理普通值和等待值的混合，我需要弄清楚哪些需要 await以产生实际值。

所以这是我的问题，什么满足 isinstance(c, ???) == True对于任意基于生成器的协程 c ?我不坚持使用 isinstance为此，也许 getattr()可以是一个解决方案...

背景

我正在开发一个基于 https://blog.miguelgrinberg.com/post/unit-testing-asyncio-code 的用于异步函数单元测试的小型模拟实用程序其中内部有一个 asyncio.Queue()模拟返回值，我想增强我的实用程序，以便队列可以有等待元素，每个元素触发 await手术。我的代码看起来像

async def case01(loop):
  f = AsyncMock()
  f.side_effect = loop, []
  await f()  # blocks forever on an empty queue

async def case02(loop):
  f = AsyncMock()
  f.side_effect = loop, ['foo']
  await f() # => 'foo'
  await f() # blocks forever

async def case03(loop):
  f = AsyncMock()
  f.side_effect = loop, [asyncio.sleep(1.0, 'bar', loop=loop)]
  await f() # yields 'bar' after 1.0 sec of delay

出于可用性原因，我不想用 create_task() 手动包装返回值。 .

我不确定我的队列是否会合法地包含正常的非协程生成器；尽管如此，理想的解决方案应该能够区分普通生成器和基于生成器的协程，并跳过应用 await操作为前者。

Answer 1

我不确定您要在这里测试什么，但是 the inspect module具有用于检查大多数此类内容的功能:

>>> async def f(c):
...     await c
>>> co = f()
>>> inspect.iscoroutinefunction(f)
True
>>> inspect.iscoroutine(co)
True
>>> inspect.isawaitable(co)
True

最后两个的区别在于isawaitable对任何事情都是如此await ，不仅仅是协程。

如果你真的想用 isinstance 测试:
isinstance(f)只是 types.FunctionType ，这不是很有用。检查它是否是一个返回 coroutine 的函数，您还需要检查其标志:f.__code__.co_flags & inspect.CO_COROUTINE (或者，如果您出于某种原因不想使用 inspect，您可以硬编码 0x80)。
isinstance(co)是 types.CoroutineType ，您可以对其进行测试，但这可能仍然不是一个好主意。