重载另一个类属性的类的类型

Question

我试图创建一个类，该类使用__getattr__调用另一个类属性，以便包装类调用

from aiohttp import ClientSession
from contextlib import asynccontextmanager


class SessionThrottler:

    def __init__(self, session: ClientSession,
                 time_period: int, max_tasks: int):
        self._obj = session
        self._bucket = AsyncLeakyBucket(max_tasks=max_tasks,
                                        time_period=time_period)

    def __getattr__(self, name):
        @asynccontextmanager
        async def _do(*args, **kwargs):
            async with self._bucket:
                res = await getattr(self._obj, name)(*args, **kwargs)
                yield res
        return _do

    async def close(self):
        await self._obj.close()

那么我可以做：

async def fetch(session: ClientSession):
    async with session.get('http://localhost:5051') as resp:
        _ = resp


session = ClientSession()
session_throttled = SessionThrottler(session, 4, 2)
await asyncio.gather(
    *[fetch(session_trottled) 
      for _ in range(10)]
)

这段代码工作得很好，但是我如何才能将session_throttled推断为ClientSession而不是SessionThrottler（有点像functools.wraps）

Answer 1

此解决方案基于修补提供给上下文管理器的对象方法（而不是包装它们）

import asyncio
import functools
import contextlib

class Counter:
    async def infinite(self):
        cnt = 0
        while True:
            yield cnt
            cnt += 1
            await asyncio.sleep(1)

def limited_infinite(f, limit):
    @functools.wraps(f)
    async def inner(*a, **kw):
        cnt = 0
        async for res in f(*a, **kw):
            yield res
            if cnt == limit:
                break
            cnt += 1
    return inner

@contextlib.contextmanager
def throttler(limit, counter):
    orig = counter.infinite
    counter.infinite = limited_infinite(counter.infinite, limit)
    yield counter
    counter.infinite = orig

async def main():
    with throttler(5, Counter()) as counter:
        async for x in counter.infinite():
            print('res: ', x)

if __name__ == "__main__":
    asyncio.run(main())

对于您的情况，它意味着修补ClientSession的每个相关方法（可能仅限于http方法）。但不确定是否更好

Answer 2

我取决于你需要什么“被推断为”

创建客户端会话的ThrotledSessions实例

用类实现这一点的自然方法是通过继承——如果你的SessionThrotler继承自ClientSession，那么它自然也是beaClientSession。 “小缺点”是__getattr__将无法按预期工作，因为只对实例中找不到的属性调用了__getattr__，Python将在ThrottledSession对象中“看到”来自ClientSession的原始方法，并调用它们

当然，这也需要静态继承类，并且您可能希望它动态工作。（我指的是静态的 必须写入class SessionThrotler(ClientSession):——或者至少，如果要包装的不同会话类数量有限，那么也要为每个会话类编写继承自ThrottledClass的子类：

class ThrotledClientSession(ThrotledSession, ClientSession):
    ...

如果这对您有用，那么这就是通过创建__getattribute__而不是__getattr__来修复属性访问的问题。两者之间的区别在于__getattribte__emcompasses属性查找步骤的所有，并在查找过程中被调用。而__getattr__作为正常查找的一部分被调用（在__getattribute__的标准算法中）当所有其他算法都失败时

class SessionThrottlerMixin:

    def __init__(self, session: ClientSession,
                 time_period: int, max_tasks: int):
        self._bucket = AsyncLeakyBucket(max_tasks=max_tasks,
                                        time_period=time_period)

    def __getattribute__(self, name):
        attr = super().__getattribute__(name)
        if not name.startswith("_") or not callable(attr):
             return attr
        @asynccontextmanager
        async def _do(*args, **kwargs):
            async with self._bucket:
                res = await attr(*args, **kwargs)
                yield res
        return _do

    class ThrotledClientSession(SessionThrottlerMixin, ClientSession):
        pass

如果您从其他代码获取CLientSession实例，并且不希望或无法使用此基类替换基类，则可以在所需实例上执行此操作，方法是将__class__属性指定为： 如果ClientSession是一个普通的Python类，不从Python内置的特殊基类继承，不使用__slots__和其他一些限制，那么它就可以工作-实例被“转换”为ThrotledClientSession中间层（但您必须执行继承操作）：session.__class__ = ThrottledClientSession

以这种方式分配的类不会运行新类__init__。因为您需要创建_bucket，您可以有一个类方法来创建bucket并进行替换-因此，在具有__getattribute__的版本中添加如下内容：

class SessionThrottler:
    ...
    @classmethod
    def _wrap(cls, instance, time_period: int, max_tasks: int):
       cls.__class__ = cls
       instance._bucket = AsyncLeakyBucket(max_tasks=max_tasks,
                                            time_period=time_period)
       return instance 

    ...

throtled_session = ThrotledClientSession._wrap(session, 4, 2)

如果您有很多父类希望以这种方式包装，并且您不想声明它的Throttled版本，那么可以动态地创建这个，但是如果这是唯一的方法，我只会这样做。最好声明10个stub-thotited版本，每个版本3行

虚拟子类化

如果您可以更改ClientSession类（以及您想要包装的其他类）的代码，这是最不显眼的方法-

Python有一个名为Virtual Subclassing的模糊OOP特性，其中一个类可以注册为另一个类的子类，而不需要真正的继承。然而，要成为“父”的类必须有abc.ABCMeta作为它的元类，否则这是非常不引人注目的

以下是它的工作原理：

In [13]: from abc import ABC                                                                                                         

In [14]: class A(ABC): 
    ...:     pass 
    ...:                                                                                                                             

In [15]: class B:  # don't inherit
    ...:     pass

In [16]: A.register(B)                                                                                                               
Out[16]: __main__.B

In [17]: isinstance(B(), A)                                                                                                          
Out[17]: True

因此，在原始代码中，如果您可以使ClientSession从abc.ABC继承（没有任何其他更改）-然后执行以下操作：

ClientSession.register(SessionThrottler) 而且它只会起作用（如果您的意思是“推断为”与对象类型有关）

请注意，如果ClientSession和其他人有不同的元类，添加abc.ABC作为其基础之一将失败，并导致元类冲突。如果您可以更改其代码，这仍然是更好的方法：只需创建一个从两个元类继承的协作元类，您就可以：

class Session(metaclass=IDontCare):
    ...


from abc import ABCMeta

class ColaborativeMeta(ABCMeta, Session.__class__):
    pass

class ClientSession(Session, metaclass=ColaborativeMeta):
    ...

类型暗示

如果您不需要“isinstance”来工作，并且只需要与键入系统相同，那么只需使用typing.cast：

import typing as T
...
session = ClientSession()
session_throttled = T.cast(ClientSession, SessionThrottler(session, 4, 2))

对象在运行时未被触碰——只是相同的对象，但是从那个点开始，像^ {CD30>}之类的工具会认为它是^ {CD2>}的实例。

最后，但并非最不重要的一点是更改类名

因此，如果“推断为”并不意味着应该将包装的类视为实例，而只关心在日志中正确显示的类名，那么只需将class __name__属性设置为所需的任何字符串：

class SessionThrottler:
    ...

SessionThrottelr.__name__ = ClientSession.__name__

或者在包装器类上使用适当的__repr__方法：

class SessionThrottler:
    ...
    def __repr__(self):
        return repr(self._obj)