根据是否实例化异步或同步实例返回包装器

Question

在给定基类的方法的情况下，有没有一种方法可以提取实例化了哪个子类？

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

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。我的 Sync 和 Async 类继承自哪个。

有什么方法可以让我知道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（或者您可以控制它们并可以添加像_is_sync 这样的类属性，告诉包装器它正在处理哪种类），您可以简单地检查self。

这不能从<my_problem> 位置完成，因为self 在那里尚不可用；代码必须为同步和异步情况返回一个包装器。一旦被调用，包装器必须检测异步情况并在您需要异步行为时返回实例化的async def。 （返回协程对象的同步函数在功能上等同于协程函数，就像以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 将不起作用；
• 如果最终类不受包装器作者控制，则自定义类属性将不起作用。

在这种情况下，剩下的选择是使用黑魔法来确定函数是从协程调用还是从同步函数调用。黑魔法由 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 版本中停止工作，而不会发出任何警告。但是，如果您知道自己在做什么，它会非常有用。