From aed36423852b963ffafa71bbe536de0963eacceb Mon Sep 17 00:00:00 2001 From: Martin Winks <50446230+uwinx@users.noreply.github.com> Date: Sun, 31 May 2020 19:01:28 +0400 Subject: [PATCH] feat(helpers): implement new descriptor with default value getter (#336) * feat(helpers): implement new descriptor with default value getter * perf(descriptor): use weakref refuse weak reference to a value in WeakRefDict instead of polluting instance namespace * chore(descriptor): rename descriptor class rename `DefaultProperty` to `Default` * style(fmt): lint code --- aiogram/api/client/session/base.py | 46 ++------------- aiogram/utils/helper.py | 58 ++++++++++++++++++- .../test_session/test_base_session.py | 9 +-- tests/test_utils/test_helper.py | 49 +++++++++++++++- 4 files changed, 112 insertions(+), 50 deletions(-) diff --git a/aiogram/api/client/session/base.py b/aiogram/api/client/session/base.py index 8213e4c3..379e69ac 100644 --- a/aiogram/api/client/session/base.py +++ b/aiogram/api/client/session/base.py @@ -8,6 +8,7 @@ from typing import Any, AsyncGenerator, Callable, ClassVar, Optional, Type, Type from aiogram.utils.exceptions import TelegramAPIError +from ....utils.helper import Default from ...methods import Response, TelegramMethod from ..telegram import PRODUCTION, TelegramAPIServer @@ -20,47 +21,10 @@ class BaseSession(abc.ABC): # global session timeout default_timeout: ClassVar[float] = 60.0 - _api: TelegramAPIServer - _json_loads: _JsonLoads - _json_dumps: _JsonDumps - _timeout: float - - @property - def api(self) -> TelegramAPIServer: - return getattr(self, "_api", PRODUCTION) # type: ignore - - @api.setter - def api(self, value: TelegramAPIServer) -> None: - self._api = value - - @property - def json_loads(self) -> _JsonLoads: - return getattr(self, "_json_loads", json.loads) # type: ignore - - @json_loads.setter - def json_loads(self, value: _JsonLoads) -> None: - self._json_loads = value # type: ignore - - @property - def json_dumps(self) -> _JsonDumps: - return getattr(self, "_json_dumps", json.dumps) # type: ignore - - @json_dumps.setter - def json_dumps(self, value: _JsonDumps) -> None: - self._json_dumps = value # type: ignore - - @property - def timeout(self) -> float: - return getattr(self, "_timeout", self.__class__.default_timeout) # type: ignore - - @timeout.setter - def timeout(self, value: float) -> None: - self._timeout = value - - @timeout.deleter - def timeout(self) -> None: - if hasattr(self, "_timeout"): - del self._timeout + api: Default[TelegramAPIServer] = Default(PRODUCTION) + json_loads: Default[_JsonLoads] = Default(json.loads) + json_dumps: Default[_JsonDumps] = Default(json.dumps) + timeout: Default[float] = Default(fget=lambda self: float(self.__class__.default_timeout)) @classmethod def raise_for_status(cls, response: Response[T]) -> None: diff --git a/aiogram/utils/helper.py b/aiogram/utils/helper.py index e582a4f0..57f4e76e 100644 --- a/aiogram/utils/helper.py +++ b/aiogram/utils/helper.py @@ -14,7 +14,10 @@ Example: <<< ['barItem', 'bazItem', 'fooItem', 'lorem'] """ import inspect -from typing import Any, Callable, Iterable, List, Optional, Union, cast +from typing import Any, Callable, Generic, Iterable, List, Optional, TypeVar, Union, cast +from weakref import WeakKeyDictionary + +T = TypeVar("T") PROPS_KEYS_ATTR_NAME = "_props_keys" @@ -233,3 +236,56 @@ class OrderedHelper(Helper, metaclass=OrderedHelperMeta): else: result.append(value) return result + + +class Default(Generic[T]): + """ + Descriptor that holds default value getter + + Example: + >>> class MyClass: + ... att = Default("dflt") + ... + >>> my_instance = MyClass() + >>> my_instance.att = "not dflt" + >>> my_instance.att + 'not dflt' + >>> MyClass.att + 'dflt' + >>> del my_instance.att + >>> my_instance.att + 'dflt' + >>> + + Intended to be used as a class attribute and only internally. + """ + + __slots__ = "fget", "_descriptor_instances" + + def __init__( + self, default: Optional[T] = None, *, fget: Optional[Callable[[Any], T]] = None, + ) -> None: + self.fget = fget or (lambda _: cast(T, default)) + self._descriptor_instances = WeakKeyDictionary() # type: ignore + + def __get__(self, instance: Any, owner: Any) -> T: + if instance is None: + return self.fget(instance) + + return self._descriptor_instances.get(instance, self.fget(instance)) + + def __set__(self, instance: Any, value: T) -> None: + if instance is None or isinstance(instance, type): + raise AttributeError( + "Instance cannot be class or None. Setter must be called from a class." + ) + + self._descriptor_instances[instance] = value + + def __delete__(self, instance: Any) -> None: + if instance is None or isinstance(instance, type): + raise AttributeError( + "Instance cannot be class or None. Deleter must be called from a class." + ) + + self._descriptor_instances.pop(instance, None) diff --git a/tests/test_api/test_client/test_session/test_base_session.py b/tests/test_api/test_client/test_session/test_base_session.py index 35dcfa8e..3ac1254b 100644 --- a/tests/test_api/test_client/test_session/test_base_session.py +++ b/tests/test_api/test_client/test_session/test_base_session.py @@ -49,14 +49,9 @@ class TestBaseSession: return json.dumps session.json_dumps = custom_dumps - assert session.json_dumps == custom_dumps == session._json_dumps + assert session.json_dumps == custom_dumps session.json_loads = custom_loads - assert session.json_loads == custom_loads == session._json_loads - - different_session = CustomSession() - assert all( - not hasattr(different_session, attr) for attr in ("_json_loads", "_json_dumps", "_api") - ) + assert session.json_loads == custom_loads def test_timeout(self): session = CustomSession() diff --git a/tests/test_utils/test_helper.py b/tests/test_utils/test_helper.py index 8125ef60..466db4ef 100644 --- a/tests/test_utils/test_helper.py +++ b/tests/test_utils/test_helper.py @@ -1,6 +1,6 @@ import pytest -from aiogram.utils.helper import Helper, HelperMode, Item, ListItem, OrderedHelper +from aiogram.utils.helper import Default, Helper, HelperMode, Item, ListItem, OrderedHelper class TestHelper: @@ -132,3 +132,50 @@ class TestOrderedHelper: B = ListItem() assert MyOrderedHelper.all() == ["A", "D", "C", "B"] + + +class TestDefaultDescriptor: + def test_descriptor_fs(self): + obj = type("ClassA", (), {})() + default_x_val = "some_x" + x = Default(default_x_val) + + # we can omit owner, usually it's just obj.__class__ + assert x.__get__(instance=obj, owner=None) == default_x_val + assert x.__get__(instance=obj, owner=obj.__class__) == default_x_val + + new_x_val = "new_x" + assert x.__set__(instance=obj, value=new_x_val) is None + + with pytest.raises(AttributeError) as exc: + x.__set__(instance=obj.__class__, value="will never be set") + assert "Instance cannot be class or None" in str(exc.value) + + assert x.__get__(instance=obj, owner=obj.__class__) == new_x_val + + with pytest.raises(AttributeError) as exc: + x.__delete__(instance=obj.__class__) + assert "Instance cannot be class or None" in str(exc.value) + + x.__delete__(instance=obj) + assert x.__get__(instance=obj, owner=obj.__class__) == default_x_val + + def test_init(self): + class A: + x = Default(fget=lambda a_inst: "nothing") + + assert isinstance(A.__dict__["x"], Default) + + a = A() + assert a.x == "nothing" + + x = Default("x") + assert x.__get__(None, None) == "x" + assert x.fget(None) == x.__get__(None, None) + + def test_nullability(self): + class A: + x = Default(default=None, fget=None) + + assert A.x is None + assert A().x is None