python - 根据实例化的是异步实例还是同步实例返回包装器

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属于基类。我的 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 , uni_wrapper的一部分 my_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 版本中停止工作。但如果您知道自己在做什么，它就会非常有用。