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

@@ -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:

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)

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()

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