Я зависит от того, что вам нужно, с помощью "выводится как".
Создание экземпляров ThrotledSessions для ClientSessions
Естественный способ сделать это с помощью классов - через наследование - если ваш SessionThrotler наследуется от ClientSession, естественно, будет a ClientSession. «Небольшой недостаток» заключается в том, что тогда __getattr__ не будет работать должным образом, поскольку вызывается только для атрибутов, не найденных в экземпляре, а Python будет «видеть» исходные методы из ClientSession в вашем объекте ThrotledSession и вызывать те, вместо
Конечно, это также потребует от вас статического наследования вашего класса, и вы можете захотеть, чтобы он работал динамически. (Статически я имею в виду необходимость писать class SessionThrotler(ClientSession): - или, по крайней мере, если существует конечное число различных классов Session, которые вы хотите обернуть, напишите также для каждого подкласса, унаследованного от ThrotledClass:
class ThrotledClientSession(ThrotledSession, ClientSession):
...
Если это то, что вам подходит, тогда нужно исправить доступ к атрибутам, создав __getattribute__ вместо __getattr__. Разница между ними заключается в том, что __getattribte__ охватывает все поиск атрибута выполняется и вызывается в начале поиска, в то время как __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 классом, не наследующим от специальных баз, таких как встроенные * 1095, не использующими __slots__ и некоторыми другими ограничениями - экземпляр "преобразуется" в ThrotledClientSession в полете (но вы должны выполнить наследование): session.__class__ = ThrottledClientSession.
Назначение класса таким образом не будет запускать новый класс __init__. Поскольку вам нужно создать _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 его версию, этот может быть создан динамически - но я бы только go таким образом, если бы это был единственный путь к go. Было бы предпочтительным объявить около 10 заглушек версий Thotled, по 3 строки в каждой.
Виртуальные подклассы
Если вы можете изменить код ваших ClientSession классов (и других, которые вы хотите обернуть), это наименее навязчивый способ - у
Python есть скрытая OOP функция, называемая Virtual Subclassing, в которой класс может быть зарегистрирован как подкласс другого без реального наследования. Однако класс, который должен быть «родительским», должен иметь 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 в качестве одной из его баз завершится с ошибкой конфликт метаклассов Если вы можете изменить их код, это все же лучший способ go: просто создайте совместный метакласс, который наследует от обоих метаклассов, и все настроено:
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))
Объект не затрагивается во время выполнения - точно такой же объект, но с этого момента такие инструменты, как mypy, будут считать его экземпляром ClientSession.
Последний, но не менее важный измените имя класса.
Таким образом, если под «выведено как» вы не подразумеваете, что обернутый класс должен рассматриваться как экземпляр, а просто заботитесь о правильном отображении имени класса в журналах и т. Д., Вы можете просто установить класс __name__ атрибут для любой строки, которую вы хотите:
class SessionThrottler:
...
SessionThrottelr.__name__ = ClientSession.__name__
или просто иметь соответствующий метод __repr__ в классе оболочки:
class SessionThrottler:
...
def __repr__(self):
return repr(self._obj)