如何判断函数或方法是普通函数还是异步函数？我希望我的代码能够自动支持普通或异步回调，并需要一种测试传递的函数类型的方式。
async def exampleAsyncCb():
pass
def exampleNomralCb():
pass
def isAsync(someFunc):
#do cool dynamic python stuff on the function
return True/False
async def callCallback(cb, arg):
if isAsync(cb):
await cb(arg)
else:
cb(arg)
根据传递的函数类型，它应该以正常方式运行或使用await运行。我尝试了各种方法，但不知道如何实现 isAsync() 。
inspect.iscoroutinefunction(object)
inspect.ismodule(object)
inspect.isclass(object)
inspect.ismethod(object)
inspect.isfunction(object)
inspect.getdoc(object)
inspect.getcomments(object)
inspect.getfile(object)
inspect.getmodule(object)
方法名称直观易懂，如需说明可在文档中查找。
inspect.iscoroutinefunction()和asyncio.iscoroutinefunction()之间是否存在实际差别？ - user4698348这个方法与 inspect.iscoroutinefunction() 不同，因为它对使用 @coroutine 装饰的基于生成器的协程函数返回 True。 - roganartuinspect.isawaitable-它检查略有不同但覆盖某些其他情况，并注意权衡将使答案更全面，2）关于Python 2回溯的旁注应该扩展说明async / `await'在Python 2中甚至不存在，或者完全省略。好的回答-两个建议：1）像其他答案一样提到inspect.isawaitable- 它检查略有不同但覆盖某些其他情况，并注意权衡将使答案更全面，2）关于Python 2回溯的旁注应该扩展说明async/await在Python 2中根本不存在，否则就应该完全省略。 - mtraceur与 inspect.iscoroutinefunction() 不同，因为它对使用 @coroutine 装饰的基于生成器的协程函数返回 True。 - Max Barraclough如果您不想引入另一个inspect导入，iscoroutine也可在asyncio内部使用。
import asyncio
def isAsync(someFunc):
return asyncio.iscoroutinefunction(someFunc)
asyncio.iscoroutinefunction() 函数有两个测试，首先使用 inspect.iscoroutinefunction() 进行测试，如果该测试失败，则测试该函数是否是应用了 @asyncio.coroutine 装饰器 的函数。请注意这一点！ - Martijn PietersTLDR
Use inspect.isawaitable to check if an object should be used with await. Unlike iscoroutine or iscoroutinefunction, it also works for Futures and objects that implement the __await__ method.
Detailed
The previously mentioned solutions work well for simple cases where you pass a coroutine function. However, in some cases, you may want to pass an awaitable object function that behaves like a coroutine function but is not one. Two examples of this are the Future class or a Future-like object class (a class that implements the __await__ magic method). In these cases, using iscoroutinefunction will return False, which is not what you need.
This can be easier to understand with a non-async example of passing a non-function callable as a callback:
class SmartCallback:
def __init__(self):
print('SmartCallback is not function, but can be used as function')
callCallback(SmartCallback) # Should work, right?
回到异步世界，情况类似：
class AsyncSmartCallback:
def __await__(self):
return self._coro().__await__()
async def _coro(self):
print('AsyncSmartCallback is not coroutine function, but can be used as coroutine function')
await asyncio.sleep(1)
await callCallback(AsyncSmartCallback) # Should work, but oops! iscoroutinefunction(AsyncSmartCallback) == False
解决此问题的方法不是使用iscoroutine或iscoroutinefunction，而是使用inspect.isawaitable。它可以处理已就绪的对象，因此您必须首先创建它。换句话说，我建议使用的解决方案是：
async def callCallback(cb, arg):
if callable(cb):
res = cb() # here's result of regular func or awaitable
if inspect.isawaitable(res):
res = await res # await if awaitable
return res # return final result
else:
raise ValueError('cb is not callable')
这是更普遍的（我相信也是逻辑上正确的）解决方案。
testcb测试，你将得到AsyncSmartCallback的结果，正如它应该的那样，没有任何等待。这是因为AsyncSmartCallback不是可等待对象，而是返回可等待对象的类：AsyncSmartCallback()-这是可等待对象。如果你的函数def testcb(): return AsyncSmartCallback()，那么这个对象将被等待。但我认为这没有什么问题：想象一下你传递了def testcb(): return Callback()-在这种情况下，Callback()也将被执行。 - Mikhail Gerasimovcb() 是一个阻塞同步函数，那么这将会阻塞整个事件循环，对吧？ - MikeTwo >>> async def f(): pass
...
>>> inspect.isawaitable(f)
False - makeroof 不可等待，但 f() 可以。inspect.isawaitable(f()) 为 True。 - Mikhail Gerasimov协程具有设置了COROUTINE标志的代码标志中的第7位，即位7：
>>> async def foo(): pass
>>> foo.__code__.co_flags & (1 << 7)
128 # not 0, so the flag is set.
在 inspect 模块中，常量 128 被存储为一个值：
>>> import inspect
>>> inspect.CO_COROUTINE
128
>>> foo.__code__.co_flags & inspect.CO_COROUTINE
128
inspect.iscoroutinefunction() 函数可以测试一个对象是否为函数或方法（确保存在 __code__ 属性），并检测该标志。请参见源代码。inspect.iscoroutinefunction() 是最易读的，并且在代码标志发生更改时也可以继续保持运行。>>> inspect.iscoroutinefunction(foo)
True
如果您的应用程序不知道给定函数的类型，它可能是上述任何一种，异步函数可以是协程函数或异步生成器函数。 asyncio.iscoroutinefunction（someFunc）仅检查函数是否为协程函数，对于异步生成器，您可以使用inspect.isasyncgenfunction()。下面是示例代码：
import inspect, asyncio
def isAsync(someFunc):
is_async_gen = inspect.isasyncgenfunction(someFunc)
is_coro_fn = asyncio.iscoroutinefunction(someFunc)
return is_async_gen or is_coro_fn
partial(some_async_func, ...) 这样的情况，它既不属于这两种情况。看起来启发式算法是我们能够得到的最接近的答案，因为我们试图猜测它是否返回一个可等待对象而不实际调用它。我想知道 async def f() -> Awaitable 是否会使 partial(...) 等更容易被检测出来。 - Tim Diels我在这里找不到其他答案来满足这个问题。至少在Python 3.10中：
class Foo:
async def __call__():
pass
foo = Foo()
asyncio.iscoroutinefunction(foo) # produces False.
相反，您可以进行以下测试：
asyncio.iscoroutinefunction(foo) or asyncio.iscoroutinefunction(foo.__call__)
Note that:
async def bar():
pass
asyncio.iscoroutinefunction(bar.__call__) # produces False
在我的代码中，我曾经有很多这样的写法：
if asyncio.iscoroutinefunction(foo):
await foo()
else:
foo()
处理其中任何一个的更清晰的模式可能是：
async def if_coro(result):
if asyncio.iscoroutine(result): # or inspect.iscoroutine,... and so on
return await result
else:
return result
result = await if_coro(async_func()) # result is as expected
result = await if_coro(sync_func()) # result is as expected
try:
result = await cb()
except TypeError as err:
if "can't be used in 'await' expression" in str(err):
result = cb()
else:
raise
cb是一个非异步的普通函数，它返回一个可等待对象，那么try块将会成功，并且你实际上会等待cb()的结果（它是一个可等待对象），尽管cb本身是一个普通函数。这可能不是你想要的 - 也许你只是想收集调用cb()的可等待结果，但还不想立即等待它。 - ustulation使用asyncio.iscoroutine()来判断协程，
以及使用asyncio.isfuture()来判断任务或者未来对象
import asyncio
async def task():
await asyncio.sleep(0.01)
print(1)
async def main():
t = task()
print(type(t))# <class 'coroutine'>
print(asyncio.iscoroutine(t)) # True
print(asyncio.isfuture(t)) # False
await t
async def main2():
t = asyncio.create_task(task())
print(type(t)) # <class '_asyncio.Task'>
print(asyncio.iscoroutine(t)) # False
print(asyncio.isfuture(t)) # True
await t
if __name__ == '__main__':
asyncio.run(main())
asyncio.run(main2())
True/False正在除以0！ :O - Shadow