<p>我取决于你需要什么“被推断为”</p>
<h2>创建客户端会话的ThrotledSessions实例</h2>
<p>用类实现这一点的自然方法是通过继承——如果你的<code>SessionThrotler</code>继承自<code>ClientSession</code>,那么它自然也是<em>be</em>a<code>ClientSession</code>。
“小缺点”是<code>__getattr__</code>将无法按预期工作,因为只对实例中找不到的属性调用了<code>__getattr__</code>,Python将在ThrottledSession对象中“看到”来自<code>ClientSession</code>的原始方法,并调用它们</p>
<p>当然,这也需要静态继承类,并且您可能希望它动态工作。(我指的是静态的
必须写入<code>class SessionThrotler(ClientSession):</code>——或者至少,如果要包装的不同会话类数量有限,那么也要为每个会话类编写继承自ThrottledClass的子类:</p>
<pre class="lang-py prettyprint-override"><code>class ThrotledClientSession(ThrotledSession, ClientSession):
...
</code></pre>
<p>如果这对您有用,那么这就是通过创建<code>__getattribute__</code>而不是<code>__getattr__</code>来修复属性访问的问题。两者之间的区别在于<code>__getattribte__</code>emcompasses<em>属性查找步骤的所有</em>,并在查找过程中被调用。而<code>__getattr__</code>作为正常查找的一部分被调用(在<code>__getattribute__</code>的标准算法中)当所有其他算法都失败时</p>
<pre class="lang-py prettyprint-override"><code>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
</code></pre>
<p>如果您从其他代码获取<code>CLientSession</code>实例,并且不希望或无法使用此基类替换基类,则可以在所需实例上执行此操作,方法是将<code>__class__</code>属性指定为:
如果<code>ClientSession</code>是一个普通的Python类,不从Python内置的特殊基类继承,不使用<code>__slots__</code>和其他一些限制,那么它就可以工作-实例被“转换”为<code>ThrotledClientSession</code>中间层(但您必须执行继承操作):<code>session.__class__ = ThrottledClientSession</code></p>
<p>以这种方式分配的类不会运行新类<code>__init__</code>。因为您需要创建<code>_bucket</code>,您可以有一个类方法来创建bucket并进行替换-因此,在具有<code>__getattribute__</code>的版本中添加如下内容:</p>
<pre class="lang-py prettyprint-override"><code>
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)
</code></pre>
<p>如果您有很多父类希望以这种方式包装,并且您不想声明它的<code>Throttled</code>版本,那么可以动态地创建这个<em>,但是如果这是唯一的方法,我只会这样做。最好声明10个stub-thotited版本,每个版本3行</p>
<h2>虚拟子类化</h2>
<p>如果您可以更改<code>ClientSession</code>类(以及您想要包装的其他类)的代码,这是最不显眼的方法-</p>
<p>Python有一个名为<code>Virtual Subclassing</code>的模糊OOP特性,其中一个类可以注册为另一个类的子类,而不需要真正的继承。然而,要成为“父”的类必须有<code>abc.ABCMeta</code>作为它的元类,否则这是非常不引人注目的</p>
<p>以下是它的工作原理:</p>
<pre class="lang-py prettyprint-override"><code>
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
</code></pre>
<p>因此,在原始代码中,如果您可以使<code>ClientSession</code>从<code>abc.ABC</code>继承(没有任何其他更改<em>)-然后执行以下操作:</p>
<p><code>ClientSession.register(SessionThrottler)</code>
而且它只会起作用(如果您的意思是“推断为”与对象类型有关)</p>
<p>请注意,如果ClientSession和其他人有不同的元类,添加<code>abc.ABC</code>作为其基础之一将失败,并导致元类冲突。如果您可以更改其代码,这仍然是更好的方法:只需创建一个从两个元类继承的协作元类,您就可以:</p>
<pre class="lang-py prettyprint-override"><code>
class Session(metaclass=IDontCare):
...
from abc import ABCMeta
class ColaborativeMeta(ABCMeta, Session.__class__):
pass
class ClientSession(Session, metaclass=ColaborativeMeta):
...
</code></pre>
<h2>类型暗示</h2>
<p>如果您不需要“isinstance”来工作,并且只需要与键入系统相同,那么只需使用<code>typing.cast</code>:</p>
<pre><code>import typing as T
...
session = ClientSession()
session_throttled = T.cast(ClientSession, SessionThrottler(session, 4, 2))
</code></pre>
<P>对象在运行时未被触碰——只是相同的对象,但是从那个点开始,像^ {CD30>}之类的工具会认为它是^ {CD2>}的实例。</P>
<h2>最后,但并非最不重要的一点是更改类名</h2>
<p>因此,如果“推断为”并不意味着应该将包装的类视为实例,而只关心在日志中正确显示的类名,那么只需将class <code>__name__</code>属性设置为所需的任何字符串:</p>
<pre class="lang-py prettyprint-override"><code>class SessionThrottler:
...
SessionThrottelr.__name__ = ClientSession.__name__
</code></pre>
<p>或者在包装器类上使用适当的<code>__repr__</code>方法:</p>
<pre class="lang-py prettyprint-override"><code>class SessionThrottler:
...
def __repr__(self):
return repr(self._obj)
</code></pre>