三个函数方法:
async def test1():
print(f'async fn')
def test2():
print(f'fn')
async def test1(arg1: int):
print(f'async fn {arg1}')我们要如何判断是否为协程呢?下面的方法可以判断
@staticmethod
def IsAsyncFunc(fn) -> tuple:
"""
检测函数是否为异步协程函数
"""
ret = fn()
if hasattr(ret, "send") and hasattr(ret, "__next__"):
# 可能是协程,但同步函数不能 await
return True, ret
return False, ret
@staticmethod
async def IsAsyncFn(fn) -> tuple:
"""
调用 fn 并判断返回值是否可 await
返回:
(is_async: bool, result)
"""
ret = fn()
try:
# 如果 ret 是协程 / async generator,就 await 它
result = await ret
return True, ret
except TypeError:
# 不能 await,说明是同步返回值
return False, ret如何调用?
把 要检测的函数 封装成一个零参函数再交给 判断, 或者传入 lambda: 函数名 (参数)