From: Christian Heller Date: Thu, 11 Sep 2025 16:35:50 +0000 (+0200) Subject: Straighten out interfaces/class hierarchy for Completables, better hide internals. X-Git-Url: https://plomlompom.com/repos/reset_cookie?a=commitdiff_plain;h=85b36a2b08d2b7cbf91c40bf18cdeeb5fb093bbd;p=ircplom Straighten out interfaces/class hierarchy for Completables, better hide internals. --- diff --git a/ircplom/client.py b/ircplom/client.py index af81d0e..1bf64d5 100644 --- a/ircplom/client.py +++ b/ircplom/client.py @@ -6,7 +6,8 @@ from dataclasses import dataclass, InitVar from enum import Enum, auto from getpass import getuser from threading import Thread -from typing import Any, Callable, Generic, NamedTuple, Optional, Self, TypeVar +from typing import (Any, Callable, Collection, Generic, Iterator, NamedTuple, + Optional, Self, TypeVar) from uuid import uuid4 # ourselves from ircplom.events import ( @@ -105,37 +106,66 @@ class _Dict(Dict[DictItem]): class _Completable(ABC): + _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 _CompletableStringsList(_Clearable, _Completable): +class _CompletableStringsCollection(_Completable, Collection): + _collected: Collection[str] + _completed: Optional[Collection[str]] = None - def __init__(self) -> None: - self._incomplete: list[str] = [] - self.completed: tuple[str, ...] = tuple() + @abstractmethod + def _copy_collected(self) -> Collection[str]: + pass + + def complete(self) -> None: + self._completed = self._copy_collected() + + def __len__(self) -> int: + assert self._completed is not None + return len(self._completed) + + def __contains__(self, item) -> bool: + assert self._completed is not None + assert isinstance(item, str) + return item in self._completed + + def __iter__(self) -> Iterator[str]: + assert self._completed is not None + yield from self._completed - def _on_list(self, m_name: str, value: str, complete: bool) -> None: - getattr(self._incomplete, m_name)(value) - if complete: - self.complete() + +class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection): + _collected: tuple[str, ...] = tuple() + _completed: Optional[tuple[str, ...]] = None + + def _copy_collected(self) -> tuple[str, ...]: + return tuple(self._collected) def append(self, value: str, complete=False) -> None: 'Append value to list.' - self._on_list('append', value, complete) + self._collected += (value,) + if complete: + self.complete() def remove(self, value: str, complete=False) -> None: 'Remove value from list.' - self._on_list('remove', value, complete) - - def complete(self) -> None: - self.completed = tuple(self._incomplete) + assert value in self._collected + self._collected = tuple(x for x in self._collected if x != value) + if complete: + self.complete() def clear(self) -> None: - self._incomplete.clear() - self.complete() + self._completed = None + self._collected = tuple() + + def into_set(self) -> set[str]: + 'Return as mere set.' + assert self._completed is not None + return set(self._completed) class IntoUpdateValueMixin(AutoAttrMixin): @@ -145,10 +175,10 @@ class IntoUpdateValueMixin(AutoAttrMixin): 'Return non-updating copy of self.' if isinstance(self, _Dict): return None - if isinstance(self, _CompletableStringsList): - return self.completed + if isinstance(self, _CompletableStringsOrdered): + return self._completed if isinstance(self, _CompletableTopic): - return Topic(*self.completed) + return Topic(*self._completed) for cls in [cls for cls in self.__class__.__mro__ if AutoAttrMixin not in cls.__mro__]: obj = cls() @@ -214,8 +244,8 @@ class _UpdatingCompletable(_UpdatingMixin, _Completable): self._on_update() -class _UpdatingCompletableStringsList(_UpdatingCompletable, - _CompletableStringsList): +class _UpdatingCompletableStringsOrdered(_UpdatingCompletable, + _CompletableStringsOrdered): pass @@ -314,7 +344,7 @@ class _CompletableTopic(_Completable): _who: Optional[NickUserHost] = None def __init__(self) -> None: - self.completed: tuple[str, Optional[NickUserHost]] = ('', None) + self._completed: tuple[str, Optional[NickUserHost]] = ('', None) @property def what(self) -> str: @@ -336,12 +366,12 @@ class _CompletableTopic(_Completable): self._who = NickUserHost(copy.nick, copy.user, copy.host) def complete(self) -> None: - self.completed = (('', None) if self._who is None - else (self._what, self._who)) + self._completed = (('', None) if self._who is None + else (self._what, self._who)) class _Channel: - user_ids: _CompletableStringsList + user_ids: _CompletableStringsOrdered topic: _CompletableTopic def __init__(self, @@ -460,7 +490,7 @@ class _UpdatingCompletableTopic(_UpdatingCompletable, _CompletableTopic): class _UpdatingChannel(_UpdatingMixin, _Channel): - user_ids: _UpdatingCompletableStringsList + user_ids: _UpdatingCompletableStringsOrdered topic: _UpdatingCompletableTopic @@ -523,8 +553,7 @@ class _UpdatingUsersDict(_UpdatingDict[_UpdatingUser]): class _UpdatingChannelsDict(_UpdatingDict[_UpdatingChannel]): def _of_user(self, user: _User) -> dict[str, _UpdatingChannel]: - return {k: v for k, v in self._dict.items() - if user.id_ in v.user_ids.completed} + return {k: v for k, v in self._dict.items() if user.id_ in v.user_ids} def of_user(self, user: _User) -> tuple[str, ...]: 'Return names of channels listing user as member.' @@ -544,7 +573,7 @@ class _ClientDb(_Clearable, _UpdatingMixin, SharedClientDbFields): caps: _UpdatingDict[_UpdatingServerCapability] channels: _UpdatingChannelsDict isupport: _UpdatingDict[str] - motd: _UpdatingCompletableStringsList + motd: _UpdatingCompletableStringsOrdered users: _UpdatingUsersDict def __getattribute__(self, key: str): @@ -608,8 +637,8 @@ class _CapsManager(_Clearable): def clear(self) -> None: self._dict.clear() - self._ls = _CompletableStringsList() - self._list = _CompletableStringsList() + self._ls = _CompletableStringsOrdered() + self._list = _CompletableStringsOrdered() self._list_expectations: dict[str, set[str]] = { 'ACK': set(), 'NAK': set()} @@ -646,13 +675,13 @@ class _CapsManager(_Clearable): elif target == self._list: acks = self._list_expectations['ACK'] naks = self._list_expectations['NAK'] - list_set = set(target.completed) + list_set = target.into_set() assert acks == list_set & acks assert set() == list_set & naks for key, data in [_Dict.key_val_from_eq_str(entry) - for entry in self._ls.completed]: + for entry in self._ls]: self._dict[key].data = data - self._dict[key].enabled = key in self._list.completed + self._dict[key].enabled = key in self._list return True return False diff --git a/ircplom/client_tui.py b/ircplom/client_tui.py index 9183a8f..77c66e7 100644 --- a/ircplom/client_tui.py +++ b/ircplom/client_tui.py @@ -80,14 +80,18 @@ class _UpdatingNode(AutoAttrMixin): self._set(update.key, update.value) return scope, update.value - def _get(self, key: str): + def _get(self, key: str) -> Any: return getattr(self, key) def _set(self, key: str, value) -> None: setattr(self, key, value) def _unset(self, key: str) -> None: - getattr(self, key).clear() + attr = getattr(self, key) + if isinstance(attr, tuple): + setattr(self, key, tuple()) + else: + attr.clear() def _is_set(self, key: str) -> bool: return hasattr(self, key) @@ -226,6 +230,8 @@ class _UpdatingChannel(_UpdatingNode): f'RAW:{self.topic.who} set topic: {self.topic.what}') return None assert update.key == 'user_ids' + if update.value is None: + return None assert isinstance(update.value, tuple) d = {'NUHS:joining': tuple(id_ for id_ in update.value if id_ not in self.user_ids)