class _Completable(ABC):
- _completed: Any
+ completed: Any
@abstractmethod
def complete(self) -> None:
- 'Set ._completed to "complete" value if possible of current state.'
+ 'Set .completed to "complete" value if possible of current state.'
class _CompletableStringsCollection(_Completable, Collection):
_collected: Collection[str]
- _completed: Optional[Collection[str]] = None
+ completed: Optional[Collection[str]] = None
@abstractmethod
def _copy_collected(self) -> Collection[str]:
pass
def complete(self) -> None:
- self._completed = self._copy_collected()
+ self.completed = self._copy_collected()
def __len__(self) -> int:
- assert self._completed is not None
- return len(self._completed)
+ assert self.completed is not None
+ return len(self.completed)
def __contains__(self, item) -> bool:
- assert self._completed is not None
+ assert self.completed is not None
assert isinstance(item, str)
- return item in self._completed
+ return item in self.completed
def __iter__(self) -> Iterator[str]:
- assert self._completed is not None
- yield from self._completed
+ assert self.completed is not None
+ yield from self.completed
class _CompletableStringsSet(_CompletableStringsCollection, Set):
_collected: set[str]
- _completed: Optional[set[str]] = None
+ completed: Optional[set[str]] = None
def __init__(self) -> None:
self._collected = set()
) -> None:
method = getattr(self._collected, m_name)
if on_complete:
- assert self._completed is not None
- assert (item in self._completed) == exists
+ assert self.completed is not None
+ assert (item in self.completed) == exists
assert (item in self._collected) == exists
method(item)
self.complete()
else:
- assert self._completed is None
+ assert self.completed is None
assert (item in self._collected) == exists
method(item)
def intersection(self, *others: Iterable[str]) -> set[str]:
'Compare self to other set(s).'
- assert self._completed is not None
- result = self._completed
+ assert self.completed is not None
+ result = self.completed
for other in others:
assert isinstance(other, set)
result = result.intersection(other)
class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection):
_collected: tuple[str, ...] = tuple()
- _completed: Optional[tuple[str, ...]] = tuple()
+ completed: Optional[tuple[str, ...]] = tuple()
def _copy_collected(self) -> tuple[str, ...]:
return tuple(self._collected)
self.complete()
-class IntoEndnodeUpdatesMixin(AutoAttrMixin):
- 'Provides .into_endnode_updates exportable for module-external consumers.'
- _cache: dict[tuple[str, ...], Any] = {}
-
- def into_endnode_updates(self,
- client_id: str,
- path: tuple[str, ...]
- ) -> list[tuple[tuple[str, ...], Any]]:
- 'Return path-value pairs for any update-worthy sub-elements.'
-
- def update_unless_cached(path: tuple[str, ...], val: Any):
- steps_so_far: list[str] = []
- for step in path[:-1]:
- cache_path = tuple(client_id) + tuple(steps_so_far)
- if cache_path in self._cache:
- del self._cache[cache_path]
- steps_so_far += [step]
- cache_path = tuple(client_id) + path
- if cache_path not in self._cache or val != self._cache[cache_path]:
- self._cache[cache_path] = val
- return [(path, val)]
- return []
-
- if isinstance(self, _Completable):
- if self._completed is not None:
- return update_unless_cached(path, self._completed)
- return []
-
- if isinstance(self, Dict) and not self._dict:
- return update_unless_cached(path, None)
-
- updates = []
- for key, val in (self._dict.items() if isinstance(self, Dict)
- else ((k, getattr(self, k))
- for k in self._deep_annotations())):
- p = path + (key,)
- if isinstance(val, IntoEndnodeUpdatesMixin):
- updates += val.into_endnode_updates(client_id, p)
- else:
- updates += update_unless_cached(p, val)
- return updates
-
-
-class _UpdatingMixin(IntoEndnodeUpdatesMixin):
+class _UpdatingMixin:
def __init__(self, on_update: Callable, **kwargs) -> None:
super().__init__(**kwargs)
self._on_update = on_update
-class _UpdatingAttrsMixin(_UpdatingMixin):
+class _UpdatingAttrsMixin(_UpdatingMixin, AutoAttrMixin):
def _make_attr(self, cls: Callable, key: str):
return cls(on_update=lambda *steps: self._on_update(key, *steps))
if hasattr(self, '_on_update') and key in self._deep_annotations():
self._on_update(key)
+ def attr_names(self) -> tuple[str, ...]:
+ 'Names of (deeply) annotated attributes.'
+ return tuple(self._deep_annotations().keys())
+
class _UpdatingDict(_UpdatingMixin, _Dict[DictItem]):
_create_if_none: Optional[dict[str, Any]] = None
+ def __bool__(self) -> bool:
+ return bool(self._dict)
+
def __getitem__(self, key: str) -> DictItem:
if key not in self._dict:
if self._create_if_none is not None:
class _CompletableTopic(_Completable, Topic):
- _completed: Optional[Topic] = None
+ completed: Optional[Topic] = None
def complete(self) -> None:
assert self.who is not None
copy = _NickUserHost.from_str(str(self.who))
- self._completed = Topic(self.what,
- NickUserHost(copy.nick, copy.user, copy.host))
+ self.completed = Topic(self.what,
+ NickUserHost(copy.nick, copy.user, copy.host))
class _Channel(Channel):
class _ClientDb(_Clearable, _UpdatingAttrsMixin, SharedClientDbFields):
+ _updates_cache: dict[tuple[str, ...], Any] = {}
_keep_on_clear = set(IrcConnSetup.__annotations__.keys())
caps: _UpdatingDict[_UpdatingServerCapability]
channels: _UpdatingChannelsDict
attr._create_if_none = {}
return attr
+ def into_endnode_updates(self, path: tuple[str, ...]
+ ) -> list[tuple[tuple[str, ...], Any]]:
+ 'Return path-value pairs for update-worthy (sub-)elements at path.'
+
+ def update_unless_cached(path: tuple[str, ...], val: Any
+ ) -> list[tuple[tuple[str, ...], Any]]:
+ if path not in self._updates_cache\
+ or val != self._updates_cache[path]:
+ self._updates_cache[path] = val
+ return [(path, val)]
+ return []
+
+ parent = self
+ cache_path: tuple[str, ...] = tuple()
+ for step in path[:-1]:
+ cache_path = cache_path + (step,)
+ if cache_path in self._updates_cache:
+ del self._updates_cache[cache_path]
+ parent = (parent[step] if isinstance(parent, Dict)
+ else getattr(parent, step))
+ last_step = path[-1]
+ val_at_path = ((parent[last_step] if last_step in parent.keys()
+ else None) if isinstance(parent, Dict)
+ else getattr(parent, last_step))
+ if not isinstance(val_at_path, _UpdatingMixin):
+ return update_unless_cached(path, val_at_path)
+ if isinstance(val_at_path, _Completable):
+ if val_at_path.completed is not None:
+ return update_unless_cached(path, val_at_path.completed)
+ return []
+ if isinstance(val_at_path, Dict) and not val_at_path.keys():
+ for cache_path in [p for p in self._updates_cache
+ if len(p) > len(path)
+ and p[:len(path)] == path]:
+ del self._updates_cache[cache_path]
+ return update_unless_cached(path, None)
+ if isinstance(val_at_path, Dict):
+ sub_items = val_at_path.keys()
+ else:
+ assert isinstance(val_at_path, _UpdatingAttrsMixin)
+ sub_items = val_at_path.attr_names()
+ updates = []
+ for key in sub_items:
+ p = path + (key,)
+ updates += self.into_endnode_updates(p)
+ return updates
+
def clear(self) -> None:
for key, value in [(k, v) for k, v in self._deep_annotations().items()
if k not in self._keep_on_clear]:
projects / ircplom / commitdiff