From: Christian Heller Date: Wed, 5 Nov 2025 19:41:45 +0000 (+0100) Subject: Code-style imported db_primitives classes as public. X-Git-Url: https://plomlompom.com/repos/%7B%7B%20web_path%20%7D%7D/%7B%7Bprefix%7D%7D/do_todos?a=commitdiff_plain;p=ircplom Code-style imported db_primitives classes as public. --- diff --git a/src/ircplom/client.py b/src/ircplom/client.py index 546026e..168c094 100644 --- a/src/ircplom/client.py +++ b/src/ircplom/client.py @@ -11,10 +11,9 @@ from typing import Any, Callable, Iterable, Optional, Self from uuid import UUID, uuid4 # ourselves from ircplom.db_primitives import ( - Dict, - _Clearable, _Completable, _CompletableStringsSet, _UpdatingAttrsMixin, - _UpdatingCompletable, _UpdatingCompletableStringsOrdered, - _UpdatingCompletableStringsSet,_UpdatingDict, _UpdatingMixin) + Clearable, Completable, CompletableStringsSet, Dict, UpdatingAttrsMixin, + UpdatingCompletable, UpdatingCompletableStringsOrdered, + UpdatingCompletableStringsSet, UpdatingDict, UpdatingMixin) from ircplom.events import ( AffectiveEvent, CrashingException, ExceptionEvent, QueueMixin) from ircplom.irc_conn import ( @@ -159,7 +158,7 @@ class IrcConnection(BaseIrcConnection, _ClientIdMixin): client_id=self.client_id).kw(e=e) -class _CompletableTopic(_Completable, Topic): +class _CompletableTopic(Completable, Topic): completed: Optional[Topic] = None def complete(self) -> None: @@ -170,7 +169,7 @@ class _CompletableTopic(_Completable, Topic): class _Channel(Channel): - user_ids: _CompletableStringsSet + user_ids: CompletableStringsSet topic: _CompletableTopic def __init__(self, @@ -303,25 +302,25 @@ class _User(_SetNickuserhostMixin, User): self._id_ = value -class _UpdatingServerCapability(_UpdatingAttrsMixin, ServerCapability): +class _UpdatingServerCapability(UpdatingAttrsMixin, ServerCapability): pass -class _UpdatingCompletableTopic(_UpdatingCompletable, _CompletableTopic): +class _UpdatingCompletableTopic(UpdatingCompletable, _CompletableTopic): pass -class _UpdatingChannel(_UpdatingAttrsMixin, _Channel): - user_ids: _UpdatingCompletableStringsSet +class _UpdatingChannel(UpdatingAttrsMixin, _Channel): + user_ids: UpdatingCompletableStringsSet topic: _UpdatingCompletableTopic - exits: _UpdatingDict[str] + exits: UpdatingDict[str] -class _UpdatingUser(_UpdatingAttrsMixin, _User): +class _UpdatingUser(UpdatingAttrsMixin, _User): pass -class _UpdatingUsersDict(_UpdatingDict[_UpdatingUser]): +class _UpdatingUsersDict(UpdatingDict[_UpdatingUser]): _top_id: int userlen: int @@ -382,7 +381,7 @@ class _UpdatingUsersDict(_UpdatingDict[_UpdatingUser]): del self[id_] -class _UpdatingChannelsDict(_UpdatingDict[_UpdatingChannel]): +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} @@ -400,7 +399,7 @@ class _UpdatingChannelsDict(_UpdatingDict[_UpdatingChannel]): channel.remove_user(user, msg) -class _UpdatingIsupportDict(_UpdatingDict[str]): +class _UpdatingIsupportDict(UpdatingDict[str]): def __delitem__(self, key: str) -> None: if key in ISUPPORT_DEFAULTS: @@ -414,13 +413,13 @@ class _UpdatingIsupportDict(_UpdatingDict[str]): self[key] = value -class _ClientDb(_Clearable, _UpdatingAttrsMixin, SharedClientDbFields): +class _ClientDb(Clearable, UpdatingAttrsMixin, SharedClientDbFields): _updates_cache: dict[tuple[str, ...], Any] _keep_on_clear = frozenset(IrcConnSetup.__annotations__.keys()) - caps: _UpdatingDict[_UpdatingServerCapability] + caps: UpdatingDict[_UpdatingServerCapability] channels: _UpdatingChannelsDict isupport: _UpdatingIsupportDict - motd: _UpdatingCompletableStringsOrdered + motd: UpdatingCompletableStringsOrdered users: _UpdatingUsersDict def __init__(self, **kwargs) -> None: @@ -474,9 +473,9 @@ class _ClientDb(_Clearable, _UpdatingAttrsMixin, SharedClientDbFields): 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): + if not isinstance(val_at_path, UpdatingMixin): return update_unless_cached(path, val_at_path) - if isinstance(val_at_path, _Completable): + if isinstance(val_at_path, Completable): if val_at_path.completed is not None: return update_unless_cached(path, val_at_path.completed) return [] @@ -489,7 +488,7 @@ class _ClientDb(_Clearable, _UpdatingAttrsMixin, SharedClientDbFields): if isinstance(val_at_path, Dict): sub_items = val_at_path.keys() else: - assert isinstance(val_at_path, _UpdatingAttrsMixin) + assert isinstance(val_at_path, UpdatingAttrsMixin) sub_items = val_at_path.attr_names() updates = [] for key in sub_items: @@ -507,7 +506,7 @@ class _ClientDb(_Clearable, _UpdatingAttrsMixin, SharedClientDbFields): value = self._deep_annotations()[key] if hasattr(value, '__origin__'): value = value.__origin__ - if issubclass(value, _Clearable): + if issubclass(value, Clearable): getattr(self, key).clear() elif issubclass(value, str): setattr(self, key, '') @@ -532,11 +531,11 @@ class _ClientDb(_Clearable, _UpdatingAttrsMixin, SharedClientDbFields): return toks[1] -class _CapsManager(_Clearable): +class _CapsManager(Clearable): def __init__(self, sender: Callable, - caps_dict: _UpdatingDict[_UpdatingServerCapability] + caps_dict: UpdatingDict[_UpdatingServerCapability] ) -> None: self._dict = caps_dict self._send = lambda *args: sender('CAP', *args) @@ -544,8 +543,8 @@ class _CapsManager(_Clearable): def clear(self) -> None: self._dict.clear() - self._ls = _CompletableStringsSet() - self._list = _CompletableStringsSet() + self._ls = CompletableStringsSet() + self._list = CompletableStringsSet() self._list_expectations: dict[str, set[str]] = { 'ACK': set(), 'NAK': set()} diff --git a/src/ircplom/db_primitives.py b/src/ircplom/db_primitives.py index 9728e2d..1895263 100644 --- a/src/ircplom/db_primitives.py +++ b/src/ircplom/db_primitives.py @@ -24,17 +24,18 @@ class AutoAttrMixin: return super().__getattribute__(key) -class _Clearable(ABC): +class Clearable(ABC): + 'Anything that has a .clear method resetting stored data to initial state.' @abstractmethod def clear(self) -> None: - 'Zero internal knowledge.' + 'Reset internal knowledge.' DictItem = TypeVar('DictItem') -class Dict(_Clearable, Generic[DictItem]): +class Dict(Clearable, Generic[DictItem]): 'Customized dict replacement.' def __init__(self, **kwargs) -> None: @@ -66,7 +67,8 @@ class Dict(_Clearable, Generic[DictItem]): del self._dict[key] -class _Completable(ABC): +class Completable(ABC): + 'Data whose collection can be declared complete/incomplete.' completed: Any @abstractmethod @@ -74,7 +76,7 @@ class _Completable(ABC): 'Set .completed to "complete" value if possible of current state.' -class _CompletableStringsCollection(_Completable, Collection): +class _CompletableStringsCollection(Completable, Collection): _collected: Collection[str] completed: Optional[Collection[str]] = None @@ -99,7 +101,8 @@ class _CompletableStringsCollection(_Completable, Collection): yield from self.completed -class _CompletableStringsSet(_CompletableStringsCollection, Set): +class CompletableStringsSet(_CompletableStringsCollection, Set): + 'Set of strings to be declared complete/incomplete.' _collected: set[str] completed: Optional[set[str]] = None @@ -141,7 +144,7 @@ class _CompletableStringsSet(_CompletableStringsCollection, Set): return result -class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection): +class _CompletableStringsOrdered(Clearable, _CompletableStringsCollection): _collected: tuple[str, ...] = tuple() completed: Optional[tuple[str, ...]] = tuple() @@ -159,14 +162,16 @@ class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection): self.complete() -class _UpdatingMixin: +class UpdatingMixin: + 'Ensures update trigger subclasses can call on any data changes.' def __init__(self, on_update: Callable, **kwargs) -> None: super().__init__(**kwargs) self._on_update = on_update -class _UpdatingAttrsMixin(_UpdatingMixin, AutoAttrMixin): +class UpdatingAttrsMixin(UpdatingMixin, AutoAttrMixin): + 'Calls update trigger on changes to attributes.' def _make_attr(self, cls: Callable, key: str): return cls(on_update=lambda *steps: self._on_update(key, *steps)) @@ -181,7 +186,8 @@ class _UpdatingAttrsMixin(_UpdatingMixin, AutoAttrMixin): return tuple(self._deep_annotations().keys()) -class _UpdatingDict(_UpdatingMixin, Dict[DictItem]): +class UpdatingDict(UpdatingMixin, Dict[DictItem]): + 'Dict calling update trigger on changes to stored data.' _preset_init_kwargs: Optional[dict[str, Any]] = None def __bool__(self) -> bool: @@ -191,7 +197,7 @@ class _UpdatingDict(_UpdatingMixin, Dict[DictItem]): if key not in self._dict: if self._preset_init_kwargs is not None: kw = {} | self._preset_init_kwargs - if _UpdatingMixin in self._item_cls.__mro__: + if UpdatingMixin in self._item_cls.__mro__: kw |= {'on_update': lambda *steps: self._on_update(key, *steps)} self[key] = self._item_cls(**kw) @@ -210,18 +216,19 @@ class _UpdatingDict(_UpdatingMixin, Dict[DictItem]): self._on_update() -class _UpdatingCompletable(_UpdatingMixin, _Completable): +class UpdatingCompletable(UpdatingMixin, Completable): + 'Template for any updating Completable.' def complete(self) -> None: super().complete() # type: ignore self._on_update() -class _UpdatingCompletableStringsSet( - _UpdatingCompletable, _CompletableStringsSet): - pass +class UpdatingCompletableStringsSet(UpdatingCompletable, + CompletableStringsSet): + 'Clearable, updating, completable set of strings.' -class _UpdatingCompletableStringsOrdered( - _UpdatingCompletable, _CompletableStringsOrdered): - pass +class UpdatingCompletableStringsOrdered(UpdatingCompletable, + _CompletableStringsOrdered): + 'Clearable, updating, completable tuple of strings.'