python 在函数内知道它是被同步调用还是异步调用

yqkkidmi  于 2023-03-11  发布在  Python
关注(0)|答案(2)|浏览(282)

我想知道是否有人知道任何神奇的函数,可以告诉我正在运行的函数是被同步调用还是异步调用。长话短说,我想做以下事情:

def fsync():
    was_called_async() # >>> False

async fasync():
    was_called_async() # >>> True

fsync()
await fasync()

这听起来可能有点奇怪,我同意。原因是因为我试图实现一个类似于functools.singledispatchmethod的函数,但用于sync和async方法。这需要使用一个装饰器,因此需要“等待-内省”。我当前的工作方法here按预期工作:

from toolbox.cofunc import cofunc
import asyncio
import time

@cofunc
def hello():
    time.sleep(0.01)
    return "hello sync world!"

@hello.register
async def _():
    await asyncio.sleep(0.01)
    return "hello async world!"

async def main():
    print(hello())        # >>> "hello sync world!"
    print(await hello())  # >>> "hello async world!"

asyncio.run(main())

然而,它利用了inspect模块,并不是我想要依赖的东西(获取前一帧并搜索await)。如果有人知道一个替代方案,那将是可爱的。

bqujaahr

bqujaahr1#

只是没有一个好的方法来做你想做的事。
函数被“同步调用”和“异步调用”之间没有实际的区别,无论哪种方式,函数都只是被调用。
最接近差异的是调用者对返回值的处理,但即使查找await也不是一个好主意,因为

await asyncio.gather(yourfunc(), something_else())

asyncio.run(yourfunc())

将直观地被认为是“异步的”。
到目前为止,最可靠的选择是只有两个功能。

smtd7mpg

smtd7mpg2#

我认为您可以使用asyncio.get_running_loop()来确定函数是否在异步上下文中调用,因为函数在异步执行期间返回正在运行的循环,如果循环没有运行,则引发RuntimeError。

import asyncio

def is_async():
    try:
        asyncio.get_running_loop()
        return True
    except RuntimeError:
        return False

def example():
    if is_async():
        print('I was called by an async function.')
    else:
        print('I was called by a sync function.')

async def async_function():
    print('I am async. is_async():', is_async())
    example()

asyncio.run(async_function())
print()
example()

控制台输出:

I am async. is_async(): True
I was called by an async function.

I was called by a sync function.

相关问题