Python的协程类型在哪里定义的?

Where is Python's coroutine type defined?

我正在研究 Python 3.5.1,我希望能够判断一个函数是否作为协程对象返回,但我找不到协程类型的定义位置,而不是最近我我一直在使用下面的代码片段通过使用函数实例化协程来获取类型。

async def _f():
    pass
COROUTINE_TYPE = type(_f())

一定有更好的方法来做到这一点,我的问题是这个类型在哪里定义,所以我可以直接使用它?

判断函数是否为协程的最佳方法是 asyncio.iscoroutinefunction

asyncio.iscoroutinefunction(some_func)

Python的协程类型在_collections_abc, which is imported in collections.abc, which itself is imported in asyncio.coroutines中定义。

虽然不确定这比@dirn 的回答更能帮助您(这是如何判断一个函数是否是协程)。

访问协程类型的最佳方式可能是通过 types 模块:

import types

types.CoroutineType  # here it is

实际上协程类型并不是 定义的 - types.py 做的事情与你正在做的几乎相同 - 但它是标准 Python 级别访问 类型的方式。

如果您想查看类型的实际定义,请参见 Include/genobject.h and Objects/genobject.c。寻找 PyCoroWhatevercoro_whatever.

的部分

如果你使用像 pycharm 这样的东西,很容易因为自动完成而在你脑海中不知道它的情况下得到它,例如

from typing import Coroutine
isinstance(my_coroutine(), Coroutine)

输出:

<input>:1: RuntimeWarning: coroutine 'my_coroutine' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
True` using

不确定与 CoroutineType 的区别是什么,或者是否重要。


所有代码:

"""
"""

import asyncio
from asyncio.events import AbstractEventLoop

import time

from typing import Coroutine

async def expensive_io(msg_to_io_device: str = '') -> str:
    await asyncio.sleep(2.5)  # simulating expensive io
    return f'expensive io done {msg_to_io_device}'

async def my_coroutine():
    result1: str = await expensive_io()
    print(f'{result1=}')
    result2: str = await expensive_io()
    print(f'{result2=}')
    result3: str = await expensive_io()
    print(f'{result3=}')

def test1():
    """
    Single coroutine that waits a couple of times but since the event loop has nothing else to do it is forced to wait.

    Learned from this example: you *have* to await a routine that is a asyncio coroutine e.g. with the async keyword.
    This is interesting because it means that calling a async function actually creates a coroutine (in python it
    creates a generator).
    """
    # - run event loop
    event_loop: AbstractEventLoop = asyncio.get_event_loop()
    coroutines = [my_coroutine()]
    event_loop.run_until_complete(*coroutines)

if __name__ == '__main__':
    #
    test1()

    print('Success, done!\a')

查看完整示例和解释:https://github.com/brando90/ultimate-utils/blob/d104bd06135113498c84e830e91ba77f413c66cd/tutorials_for_myself/concurrency/asyncio_my_example.py#L87