1from asyncutils._internal.compat import Placeholder, partial
2from asyncutils._internal.helpers import LoopMixinBase, fullname, subscriptable
3from asyncutils._internal.submodules import properties_all as __all__
4from _collections import deque
5from weakref import WeakKeyDictionary as W
6import asyncutils as A, abc as a
[docs]
7class Deleters(__import__('enum').IntFlag): CANNOT_SET_AFTER_DELETE = SILENT = 1; DEFAULT, CAN_DELETE_AGAIN, NO_DELETER = 0, 2, 4
8d, n = Deleters.DEFAULT, 'None'
9class D(W):
10 def __init__(self, v, d=None, /): super().__init__(); self.c, self.v, self.d = bool, v, d
11 def __set_name__(self, o, n, /, _=frozenset(('__doc__', '__module__', '__name__'))):
12 if n not in _: raise TypeError('asyncutils: internal descriptor bound to forbidden name')
13 self.c = o
14 def __get__(self, o, t=None, /):
15 if not (t is None or issubclass(t, self.c)): raise TypeError('asyncutils: internal descriptor bound incorrectly')
16 return self.v if o is None else self.get(o, self.d)
17 def __str__(self, _=n, /): return _ if (v := self.v) is None else v
18 def __repr__(self, _=n, /): return _ if (v := self.v) is None else repr(v)
19 __set__, __delete__ = W.__setitem__, W.__delitem__
[docs]
20@subscriptable
21class AsyncPropertyBase(LoopMixinBase, metaclass=type('AsyncPropertyMeta', (a.ABCMeta,), {'__prepare__': classmethod(lambda c, n, b, _=D, /, **k: (p := c.__base__.__prepare__(n, b, **k)).update(__doc__=_(None), __name__=_(n)) or p), '__new__': lambda m, n, b, d, _=D, /, **k: d.__setitem__('__module__', _('asyncutils.properties')) or m.__base__.__new__(m, n, b, d, **k)})):
22 __slots__ = '__ca', '__cls', '__deleted', '__hide', '__lock', '__mutable', '__q', '__strict', '__weakref__', 'fdel', 'fget', 'fset'; _repr_accessor = repr
[docs]
23 def _init(self, f, /, fset=None, fdel=d, *, _='<unknown>', doc=None, strict=True, mutable=False, assert_modifiers_return_none=True, hide=False): self.fget, self.fset, self.fdel, self.__doc__, self.__module__, self.__deleted, self.__strict, self.__hide, self.__ca, self.__q, self.__lock, self.__mutable, self.__cls = f, fset, fdel, getattr(f, '__doc__', None) if doc is None else doc, getattr(f, '__module__', _), set(), strict, hide, assert_modifiers_return_none, deque(), type(self)._lock_factory(), mutable, None
[docs]
24 @a.abstractmethod
25 def wrap_aw(self, _, /): raise NotImplementedError
26 def __new__(cls, fget=None, *a, **k):
27 if fget is None: return partial(cls, *((Placeholder, *a) if a else ()), **k)
28 (_ := object.__new__(cls))._init(fget, *a, **k); return _
29 def __default_deleter(self, o, f, /):
30 if f&Deleters.NO_DELETER: self.__raise('asyncutils.properties.AsyncPropertyBase: undeletable attribute', o, not f&Deleters.SILENT)
31 if o in (d := self.__deleted): return self.__raise('asyncutils.properties.AsyncPropertyBase: attribute was already deleted', o, not f&Deleters.CAN_DELETE_AGAIN)
32 d.add(o)
[docs]
33 async def __get__(self, o, _=None, /):
34 if self.__check_instance(o, not (h := self.__hide)): ...
35 elif self.fget is None: self.__raise('asyncutils.properties.AsyncPropertyBase: unreadable attribute', o, self.__strict or h)
36 elif o in self.__deleted:
37 if type(d := self.fdel) is Deleters: self.__raise('asyncutils.properties.AsyncPropertyBase: attribute was deleted', o, not d&Deleters.NO_DELETER)
38 else: raise A.StateCorrupted('async property internal', 'default deleter was called on property with proper deleter')
39 else: return await self.__get(o)
40 return self
[docs]
41 def __set__(self, o, v, /):
42 if self.__check_instance(o): return
43 if (f := self.fset) is None: self.__raise('asyncutils.properties.AsyncPropertyBase: immutable attribute', o, self.__strict); return
44 if o in (s := self.__deleted): self.__raise('asyncutils.properties.AsyncPropertyBase: cannot set deleted attribute', o, type(d := self.fdel) is Deleters and d&Deleters.CANNOT_SET_AFTER_DELETE); s.discard(o)
45 self.__helper(f, 'set', o, v)
[docs]
46 def __delete__(self, o, /):
47 if not self.__check_instance(o): self.__default_deleter(o, f) if isinstance(f := self.fdel, Deleters) else self.__helper(f, 'delete', o)
[docs]
48 def __set_name__(self, /, *_): self.__cls, self.__name__ = _
49 def __repr__(self): return f'asyncutils.properties.{type(self).__name__}({', '.join(map(type(self)._repr_accessor, (self.fget, self.fset, self.fdel)))}, doc={self.__doc__!r}, strict={self.__strict}, mutable={self.__mutable}, assert_modifiers_return_none={self.__ca}, hide={self.__hide})'
[docs]
50 def __reduce__(self):
51 if self.__hide: raise TypeError('asyncutils.properties.AsyncPropertyBase: cannot pickle hidden property')
52 return f'{self.__check_unbound().__name__}.{self.__name__}'
53 def __check_unbound(self):
54 if (c := self.__cls) is None: raise TypeError(f'{self!r} is not bound to a class')
55 return c
56 def __check_instance(self, o, b=None, /):
57 c = self.__check_unbound()
58 if o is None: self.__raise('asyncutils.properties.AsyncPropertyBase.__get__ called incorrectly', c, not (self.__mutable if b is None else b)); return True
59 self.__raise('asyncutils.properties.AsyncPropertyBase.__get__ called incorrectly', o, not isinstance(o, c)); return False
60 async def __get(self, o, /):
61 p, b, r = (q := self.__q).popleft, self.__ca, self.__raise
62 async with self.__lock:
63 while q: r('asyncutils.properties.AsyncPropertyBase: setter or deleter returned non-None value', o, await p() is not None and b)
64 return await self.fget(o)
65 def __raise(self, m, o, c=True, /):
66 if c: raise AttributeError(m, name=self.__name__, obj=o) from None
67 def __helper(self, f, c, /, *a):
68 if (r := f(*a)) is None: return
69 try: self.__q.append(self.wrap_aw(r))
70 except A.CRITICAL: raise A.Critical
71 except TypeError:
72 if self.__ca: raise
73 except BaseException as e: self.__raise(f'failed to {c} attribute due to {fullname(e)}: {e}', a[0]) # noqa: BLE001
[docs]
74 def getter(self, f, /): return type(self)(f, self.fset, self.fdel, doc=self.__doc__, strict=self.__strict, mutable=self.__mutable, assert_modifiers_return_none=self.__ca, hide=self.__hide)
[docs]
75 def setter(self, f, /): return type(self)(self.fget, f, self.fdel, doc=self.__doc__, strict=self.__strict, mutable=self.__mutable, assert_modifiers_return_none=self.__ca, hide=self.__hide)
[docs]
76 def deleter(self, f, /): return type(self)(self.fget, self.fset, f, doc=self.__doc__, strict=self.__strict, mutable=self.__mutable, assert_modifiers_return_none=self.__ca, hide=self.__hide)
[docs]
77 def __getattr__(self, n, /):
78 if (f := self.fget) is None: raise AttributeError('asyncutils.properties.AsyncPropertyBase: property has no getter to find attribute on', name=n, obj=self)
79 return getattr(f, n)
[docs]
80 def __init_subclass__(cls, /, *, lock_factory=None, **k):
81 if not isinstance(cls.__dict__.get('__slots__'), tuple): raise TypeError('subclass of asyncutils.properties.AsyncPropertyBase must define tuple __slots__')
82 if lock_factory is not None: cls._lock_factory = lock_factory
83 elif getattr(cls, '_lock_factory', None) is None: raise TypeError('asyncutils.properties.AsyncPropertyBase subclasses must specify lock_factory')
84 super().__init_subclass__(**k)
[docs]
85class LazyAsyncProperty(AsyncPropertyBase, lock_factory=__import__('asyncio').Lock): __slots__, wrap_aw = (), staticmethod(A.wrap_in_coro)
[docs]
86class ConcurrentAsyncProperty(AsyncPropertyBase, lock_factory=lambda _=A.anullcontext, /: _): __slots__, wrap_aw = (), LoopMixinBase.make
[docs]
87class RWLockedAsyncProperty(ConcurrentAsyncProperty):
88 __slots__, _repr_accessor = (), lambda v, _=n, /: _ if v is None else repr(v) if type(v) is Deleters else repr(v.__wrapped__)
[docs]
89 def _init(self, f, /, fset=None, fdel=d, *, policy=A.RWLock, **k):
90 if not issubclass(policy, A.RWLock): raise TypeError('asyncutils.properties.RWLockedAsyncProperty: policy must be a subclass of asyncutils.rwlocks.RWLock')
91 w = (f := policy.lock(f)).writer; super()._init(f, None if fset is None else w(fset), fdel if fdel is Deleters.DEFAULT else w(fdel), **k)
92del a, d, n, W, D