Python inspect 怎样判断函数是不是 coroutine

判断一个对象是不是 coroutine function，直接使用 inspect.iscoroutinefunction(obj)。它检查的是“函数对象本身”：用 async def 定义的函数、绑定方法，以及包装协程函数的 functools.partial 都可以进入这个判断。已经调用后的结果则要改用 inspect.iscoroutine(obj)，不能把两个问题混在一起。

官方地址：https://docs.python.org/3.14/library/inspect.html

最稳妥的边界是：判断调用能力用 iscoroutinefunction，判断已经产生的运行时对象用 iscoroutine 或 isawaitable。同步函数如果只是“返回一个协程”，默认不会因为返回值而自动变成 coroutine function。

下面按“保护判断边界”的方式处理：先确定检查对象，再识别可接受的函数形态，最后单独处理同步包装和版本差异。

先分清要检查的是函数还是调用结果

协程函数和 coroutine 对象是两个层次。async def fetch() 产生的是函数对象；调用 fetch() 后才会得到 coroutine 对象。前者适合交给 inspect.iscoroutinefunction，后者适合交给 inspect.iscoroutine。异步生成器函数还有独立的 inspect.isasyncgenfunction 判断，不能把“带 async”理解成同一类对象。

import inspect


async def fetch_user():
    # async def 定义的是协程函数，调用后才创建 coroutine 对象
    return {"name": "Ada"}


def regular_worker():
    # 普通函数不是协程函数，即使名字或返回值另有约定也不改变定义形态
    return "done"


async def stream_rows():
    # async generator function 使用 yield，应该单独识别
    yield 1


coroutine_obj = fetch_user()
try:
    print(inspect.iscoroutinefunction(fetch_user))
    print(inspect.iscoroutinefunction(regular_worker))
    print(inspect.iscoroutine(coroutine_obj))
    print(inspect.isasyncgenfunction(stream_rows))
finally:
    # 这里只关闭示例中尚未交给事件循环的 coroutine，避免资源警告
    coroutine_obj.close()

这个区分还有一个实际安全意义：如果把函数误调用后再传给“函数检查”代码，判断结果就会失去原本的语义；如果为了检查而随意调用函数，还可能触发网络请求、文件访问或创建未等待的 coroutine。检查函数对象通常不需要执行被检查函数。

用 iscoroutinefunction 做函数类型判断

Python 3.14 文档将 inspect.iscoroutinefunction 定义为：对象是 async def 定义的协程函数、包装协程函数的 functools.partial，或者被 markcoroutinefunction() 标记的同步函数时返回 True。绑定到实例上的异步方法也属于协程函数判断的正常输入。

import functools
import inspect


async def query_user(user_id):
    # 这是最直接、也最推荐的协程函数定义
    return user_id


class Client:
    async def load(self, key):
        # 绑定方法访问时仍然可以按协程函数检查
        return key


partial_query = functools.partial(query_user, 7)
client = Client()

print(inspect.iscoroutinefunction(query_user))
print(inspect.iscoroutinefunction(client.load))
print(inspect.iscoroutinefunction(partial_query))

这里的判断不等于“调用一定成功”，也不等于“调用后一定已经执行”。它只回答函数形态问题。真正调用时仍然需要 await、任务调度或其他明确的生命周期管理。

同步函数返回协程时，怎样处理兼容边界

有些适配器必须保留同步函数外形，但内部返回一个 coroutine。此时可以在 Python 3.12 及更高版本使用 inspect.markcoroutinefunction，让依赖 iscoroutinefunction 的接口知道这个 callable 的协程语义。官方文档同时建议：能写成 async def 时优先直接使用它；另一种可靠办法是调用后用 inspect.iscoroutine 检查返回对象。

import inspect


async def build_result(value):
    # 真正的异步工作放在协程函数中
    return value


@inspect.markcoroutinefunction
def adapter(value):
    # 适配器保持同步定义，但明确返回 coroutine
    return build_result(value)


print(inspect.iscoroutinefunction(adapter))

result = adapter("ok")
try:
    print(inspect.iscoroutine(result))
finally:
    # 示例不负责调度该对象，因此显式关闭它
    result.close()

markcoroutinefunction 是给“函数检测协议”使用的标记，不会把同步函数凭空变成异步执行器，也不会替你等待返回值。若适配器有时返回普通值、有时返回 coroutine，就不应只凭装饰器隐藏这个差异，调用方应按实际返回对象设计统一协议。

按版本和对象类型排查误判

遇到结果与预期不一致时，可以按下面的边界清单排查：

版本差异也要纳入判断：partial 包装协程函数的识别在 Python 3.8 起得到支持；同步函数的 markcoroutinefunction 标记在 Python 3.12 加入；partialmethod 包装协程函数的识别在 Python 3.13 起得到支持。如果项目需要兼容更早版本，不要把新版本行为当成所有运行环境的共同能力。

常见问题

调用 async 函数后，应该用哪个函数判断？

用 inspect.iscoroutine(result) 判断调用结果；如果要判断更宽的可等待对象，使用 inspect.isawaitable(result)。

普通函数返回 coroutine，iscoroutinefunction 会自动返回 True 吗？

不会。默认判断关注函数本身；可以改用 async def，或者在 Python 3.12+ 使用 markcoroutinefunction 明确标记，再由调用方负责正确调度返回的 coroutine。

为什么 async generator 不用 iscoroutinefunction 判断？

它是另一种异步函数形态，使用 inspect.isasyncgenfunction 更准确；调用后得到的是异步生成器对象，也应按对应协议消费。

参考资料：https://docs.python.org/3.14/library/inspect.html#inspect.iscoroutinefunction；https://docs.python.org/3.14/library/inspect.html#inspect.markcoroutinefunction；https://docs.python.org/3.14/library/asyncio-task.html#asyncio-coroutines。