mirror of
https://github.com/aiogram/aiogram.git
synced 2026-04-08 16:37:47 +00:00
perf(descriptor): use weakref
refuse weak reference to a value in WeakRefDict instead of polluting instance namespace
This commit is contained in:
parent
fb63619e52
commit
a2e85c5a42
3 changed files with 21 additions and 78 deletions
|
|
@ -15,6 +15,7 @@ Example:
|
|||
"""
|
||||
import inspect
|
||||
from typing import Any, Callable, Generic, Iterable, List, Optional, TypeVar, Union, cast
|
||||
from weakref import WeakKeyDictionary
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
|
@ -239,56 +240,37 @@ class OrderedHelper(Helper, metaclass=OrderedHelperMeta):
|
|||
|
||||
class DefaultProperty(Generic[T]):
|
||||
"""
|
||||
Implements descriptor. Intended to be used as a class attribute and only internally.
|
||||
Descriptor that holds default value and for a class and
|
||||
|
||||
Intended to be used as a class attribute and only internally.
|
||||
"""
|
||||
|
||||
__slots__ = (
|
||||
"name_resolver",
|
||||
"name",
|
||||
"fget",
|
||||
)
|
||||
__slots__ = "fget", "_descriptor_instances"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
default: Optional[T] = None,
|
||||
*,
|
||||
name_resolver: Callable[[str], str] = lambda s: "_" + s,
|
||||
fget: Optional[Callable[[Any], T]] = None,
|
||||
self, default: Optional[T] = None, *, fget: Optional[Callable[[Any], T]] = None,
|
||||
) -> None:
|
||||
if fget is None is default:
|
||||
raise ValueError("Either default or fget should be passed.")
|
||||
|
||||
self.fget = fget or (lambda _: cast(T, default))
|
||||
self.name_resolver = name_resolver
|
||||
self.name: Optional[str] = None
|
||||
|
||||
def __set_name__(self, owner: Any, name: str) -> None:
|
||||
self.name = self.name_resolver(name)
|
||||
|
||||
def _raise_if_name_never_set(self, instance: Any) -> None:
|
||||
if self.name is None:
|
||||
raise AttributeError(f"Name for descriptor was never set in {instance}")
|
||||
self._descriptor_instances = WeakKeyDictionary() # type: ignore
|
||||
|
||||
def __get__(self, instance: Any, owner: Any) -> T:
|
||||
if instance is None:
|
||||
return self.fget(instance)
|
||||
|
||||
self._raise_if_name_never_set(instance)
|
||||
|
||||
return cast(T, getattr(instance, self.name, self.fget(instance))) # type: ignore
|
||||
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._raise_if_name_never_set(instance)
|
||||
setattr(instance, self.name, value) # type: ignore
|
||||
|
||||
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._raise_if_name_never_set(instance)
|
||||
delattr(instance, self.name) # type: ignore
|
||||
|
||||
self._descriptor_instances.pop(instance, None)
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -139,7 +139,6 @@ class TestDefaultDescriptor:
|
|||
obj = type("ClassA", (), {})()
|
||||
default_x_val = "some_x"
|
||||
x = DefaultProperty(default_x_val)
|
||||
x.__set_name__(owner=obj.__class__, name="x")
|
||||
|
||||
# we can omit owner, usually it's just obj.__class__
|
||||
assert x.__get__(instance=obj, owner=None) == default_x_val
|
||||
|
|
@ -161,45 +160,12 @@ class TestDefaultDescriptor:
|
|||
x.__delete__(instance=obj)
|
||||
assert x.__get__(instance=obj, owner=obj.__class__) == default_x_val
|
||||
|
||||
def test_set_name(self):
|
||||
class A:
|
||||
x = DefaultProperty("default")
|
||||
|
||||
a = A()
|
||||
assert "_x" not in vars(a)
|
||||
|
||||
a.x = "new value"
|
||||
assert "_x" in vars(a)
|
||||
|
||||
del a.x
|
||||
assert "_x" not in vars(a)
|
||||
|
||||
assert a.x == "default"
|
||||
|
||||
class B:
|
||||
x = DefaultProperty("default", name_resolver=lambda name: f"_{name}_4_{name}_")
|
||||
|
||||
b = B()
|
||||
b.x = ">>"
|
||||
assert "_x_4_x_" in vars(b)
|
||||
|
||||
C = type("C", (), {"x": DefaultProperty("default")})
|
||||
c = C()
|
||||
assert c.x == "default"
|
||||
c.x = "new value"
|
||||
assert "_x" in vars(c)
|
||||
|
||||
d = type("D", (), {})()
|
||||
x = DefaultProperty("default")
|
||||
|
||||
with pytest.raises(AttributeError) as exc:
|
||||
x.__set__(d, "new_value")
|
||||
assert f"Name for descriptor was never set in {d}" in str(exc.value)
|
||||
|
||||
def test_init(self):
|
||||
class A:
|
||||
x = DefaultProperty(fget=lambda a_inst: "nothing")
|
||||
|
||||
assert isinstance(A.__dict__["x"], DefaultProperty)
|
||||
|
||||
a = A()
|
||||
assert a.x == "nothing"
|
||||
|
||||
|
|
@ -207,9 +173,9 @@ class TestDefaultDescriptor:
|
|||
assert x.__get__(None, None) == "x"
|
||||
assert x.fget(None) == x.__get__(None, None)
|
||||
|
||||
with pytest.raises(ValueError) as exc:
|
||||
def test_nullability(self):
|
||||
class A:
|
||||
x = DefaultProperty(default=None, fget=None)
|
||||
|
||||
class _:
|
||||
x = DefaultProperty(default=None, fget=None)
|
||||
|
||||
assert "Either default or fget should be passed." in str(exc.value)
|
||||
assert A.x is None
|
||||
assert A().x is None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue