Return 一个包装器，取决于实例化的是异步实例还是同步实例

Question

有没有一种方法可以提取哪个 subclass 是根据基 class 中的方法实例化的？

我知道问题有点复杂，举个例子：

from functools import wraps

def my_wrapper(fn_to_wrap):

    @wraps(fn_to_wrap)
    async def async_wrapper(*args, **kwargs):
        await do_some_async_stuff()
        print('Did some async stuff')
        return fn_to_wrap(*args, **kwargs)

    @wraps(fn_to_wrap)
    def sync_wrapper(*args, **kwargs):
        do some_sync_stuff()
        print('Did some sync stuff')
        return fn_to_wrap(*args, **kwargs)

    # <my_problem>
    if fn_to_wrap belongs_to(SyncClass):
        return sync_wrapper
    else:
        return async_wrapper
    # </my_problem>


class BaseClass:
    @my_wrapper
    def fn_to_wrap(self):
        return 'Finally a return a value'

class SyncClass(BaseClass):
    def fn_to_call(self):
        return self.fn_to_wrap()


class AsyncClass(BaseClass):
    async def fn_to_call(self):
        return await self.fn_to_wrap()

问题是方法fn_to_wrap属于BaseClass。我的同步和异步 classes 继承自。

有什么方法可以知道 fn_to_wrap 属于 AsyncClass 还是 SyncClass 的实例？

简单地说，我希望我的控制台打印：

>>> my_sync_class = SyncClass()
>>> print(my_sync_class.fn_to_call())

Done some sync stuff
Finally a return value

和

>>> my_async_class = AsyncClass()
# not in a coroutine for brevity
>>> print(await my_async_class.fn_to_call())

Done some async stuff
FInally a return value

那么，您将如何实施 </my_problem> 来实现这些结果？

[编辑]

我知道 inspect.iscoroutinefunction 和 inspect.iscoroutine 的存在。但这些都无济于事，因为包装方法始终是同步的，而包装器是执行异步任务的方法。

Answer 1

如果 my_wrapper 被允许知道 AsyncClass 和 SyncClass（或者你控制它们并可以添加一个 class 属性，比如 _is_sync 告诉它处理的是哪种 class 的包装器），你可以简单地检查 self.

这不能从 <my_problem> 位置完成，因为 self 在那里还不可用；代码必须 return 一个用于同步和异步情况的包装器。一旦调用，包装器必须检测异步情况和 return 实例化 async def 如果您需要异步行为。（同步函数 returns 协程对象在功能上等同于协程函数，很像以 return some_generator() 结尾的普通函数，完全可以用作发电机。）

这是一个使用 isinstance 检测调用了哪个变体的示例：

def my_wrapper(fn_to_wrap):
    async def async_wrapper(*args, **kwargs):
        await asyncio.sleep(.1)
        print('Did some async stuff')
        return fn_to_wrap(*args, **kwargs)

    @wraps(fn_to_wrap)
    def uni_wrapper(self, *args, **kwargs):
        # or if self._is_async, etc.
        if isinstance(self, AsyncClass):
            return async_wrapper(self, *args, **kwargs)
        time.sleep(.1)
        print('Did some sync stuff')
        return fn_to_wrap(self, *args, **kwargs)

    return uni_wrapper

该实施会产生所需的输出：

>>> x = SyncClass()
>>> x.fn_to_call()
Did some sync stuff
'Finally a return a value'
>>> async def test():
...     x = AsyncClass()
...     return await x.fn_to_call()
... 
>>> asyncio.get_event_loop().run_until_complete(test())
Did some async stuff
'Finally a return a value'

如果包装器无法区分 SyncClass 和 AsyncClass，则上述解决方案将不起作用。有两个约束可能会阻止它这样做：

isinstance如果subclasses的数量是开放式的就不行了；
如果最终 classes 不受包装器作者控制，自定义 class 属性将不起作用。

在那种情况下，剩下的选择就是诉诸黑魔法来确定该函数是从协程还是从同步函数调用的。黑魔法由 David Beazley 在 this talk:

中方便地提供

def from_coroutine():
    return sys._getframe(2).f_code.co_flags & 0x380

使用 from_coroutine，my_wrapper 的 uni_wrapper 部分将如下所示：

    @wraps(fn_to_wrap)
    def uni_wrapper(*args, **kwargs):
        if from_coroutine():
            return async_wrapper(*args, **kwargs)
        time.sleep(.1)
        print('Did some sync stuff')
        return fn_to_wrap(*args, **kwargs)

...提供相同的结果。

当然，您必须知道黑魔法可能会在下一个 Python 版本中停止工作，而不会发出任何警告。但是，如果您知道自己在做什么，它会非常有用。

Return 一个包装器，取决于实例化的是异步实例还是同步实例

Return a wrapper depending on whether an async or sync instance was instantiated

python

wrapper

python-3.x

async-await

python-asyncio