From: Christian Heller Date: Wed, 5 Nov 2025 19:05:02 +0000 (+0100) Subject: Move data structure primitives out of client.py into new db_primitives.py. X-Git-Url: https://plomlompom.com/repos/%7B%7Bdb.prefix%7D%7D/%7B%7B%20web_path%20%7D%7D/day_todos?a=commitdiff_plain;h=ae0d85aa7a3930167e09520aec6adb586ae72ce3;p=ircplom Move data structure primitives out of client.py into new db_primitives.py. --- diff --git a/src/ircplom/client.py b/src/ircplom/client.py index 9b66efd..5a138dd 100644 --- a/src/ircplom/client.py +++ b/src/ircplom/client.py @@ -7,10 +7,15 @@ from getpass import getuser from re import search as re_search, IGNORECASE as re_IGNORECASE from threading import Thread from time import sleep -from typing import (Any, Callable, Collection, Generic, Iterable, Iterator, - Optional, Self, Set, TypeVar) +from typing import Any, Callable, Iterable, Optional, Self from uuid import UUID, uuid4 # ourselves +from ircplom.db_primitives import ( + Dict, + _Clearable, _Completable, _CompletableStringsSet, _Dict, + _UpdatingAttrsMixin, _UpdatingCompletable, + _UpdatingCompletableStringsOrdered, _UpdatingCompletableStringsSet, + _UpdatingDict, _UpdatingMixin,) from ircplom.events import ( AffectiveEvent, CrashingException, ExceptionEvent, QueueMixin) from ircplom.irc_conn import ( @@ -38,241 +43,6 @@ class ImplementationFail(Exception): 'When no matching parser found for server message.' -class AutoAttrMixin: - 'Ensures attribute as defined by annotations along MRO' - - @classmethod - def _deep_annotations(cls) -> dict[str, type]: - types: dict[str, type] = {} - for c in cls.__mro__: - if hasattr(c, '__annotations__'): - types = c.__annotations__ | types - return {k: v for k, v in types.items() if k[0] != '_'} - - def __getattribute__(self, key: str): - if key[0] != '_' and (cls := self._deep_annotations().get(key, None)): - try: - return super().__getattribute__(key) - except AttributeError: - setattr(self, key, self._make_attr(cls, key)) - return super().__getattribute__(key) - - -class _Clearable(ABC): - - @abstractmethod - def clear(self) -> None: - 'Zero internal knowledge.' - - -DictItem = TypeVar('DictItem') - - -class Dict(_Clearable, Generic[DictItem]): - 'Customized dict replacement.' - - def __init__(self, **kwargs) -> None: - self._dict: dict[str, DictItem] = {} - super().__init__(**kwargs) - - def keys(self) -> tuple[str, ...]: - 'Keys of item registrations.' - return tuple(self._dict.keys()) - - def __getitem__(self, key: str) -> DictItem: - return self._dict[key] - - def clear(self) -> None: - self._dict.clear() - - @property - def _item_cls(self): - orig_cls = (self.__orig_class__ if hasattr(self, '__orig_class__') - else self.__orig_bases__[0]) - return orig_cls.__args__[0] - - -class _Dict(Dict[DictItem]): - - def __setitem__(self, key: str, val: DictItem) -> None: - assert isinstance(val, self._item_cls) - self._dict[key] = val - - def __delitem__(self, key: str) -> None: - del self._dict[key] - - def values(self) -> tuple[DictItem, ...]: - 'Items registered.' - return tuple(self._dict.values()) - - @staticmethod - def key_val_from_eq_str(eq_str: str) -> tuple[str, str]: - 'Split eq_str by "=", and if none, into eq_str and "".' - toks = eq_str.split('=', maxsplit=1) - return toks[0], '' if len(toks) == 1 else toks[1] - - -class _Completable(ABC): - completed: Any - - @abstractmethod - def complete(self) -> None: - 'Set .completed to "complete" value if possible of current state.' - - -class _CompletableStringsCollection(_Completable, Collection): - _collected: Collection[str] - completed: Optional[Collection[str]] = None - - @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 - - -class _CompletableStringsSet(_CompletableStringsCollection, Set): - _collected: set[str] - completed: Optional[set[str]] = None - - def __init__(self) -> None: - self._collected = set() - - def _copy_collected(self) -> set[str]: - return set(self._collected) - - def _on_set(self, m_name: str, item: str, on_complete: bool, exists: bool - ) -> None: - method = getattr(self._collected, m_name) - if on_complete: - 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 (item in self._collected) == exists - method(item) - - def completable_add(self, item: str, on_complete: bool) -> None: - 'Put item into collection.' - self._on_set('add', item, on_complete, False) - - def completable_remove(self, item: str, on_complete: bool) -> None: - 'Remove item from collection.' - self._on_set('remove', item, on_complete, True) - - def intersection(self, *others: Iterable[str]) -> set[str]: - 'Compare self to other set(s).' - assert self.completed is not None - result = self.completed - for other in others: - assert isinstance(other, set) - result = result.intersection(other) - return result - - -class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection): - _collected: tuple[str, ...] = tuple() - completed: Optional[tuple[str, ...]] = tuple() - - def _copy_collected(self) -> tuple[str, ...]: - return tuple(self._collected) - - def append(self, value: str, complete=False) -> None: - 'Append value to list.' - self._collected += (value,) - if complete: - self.complete() - - def clear(self) -> None: - self._collected = tuple() - self.complete() - - -class _UpdatingMixin: - - def __init__(self, on_update: Callable, **kwargs) -> None: - super().__init__(**kwargs) - self._on_update = on_update - - -class _UpdatingAttrsMixin(_UpdatingMixin, AutoAttrMixin): - - def _make_attr(self, cls: Callable, key: str): - return cls(on_update=lambda *steps: self._on_update(key, *steps)) - - def __setattr__(self, key: str, value) -> None: - super().__setattr__(key, value) - 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: - kw = {} | self._create_if_none - if _UpdatingMixin in self._item_cls.__mro__: - kw |= {'on_update': - lambda *steps: self._on_update(key, *steps)} - self[key] = self._item_cls(**kw) - return super().__getitem__(key) - - def __setitem__(self, key: str, val: DictItem) -> None: - super().__setitem__(key, val) - self._on_update(key) - - def __delitem__(self, key: str) -> None: - super().__delitem__(key) - self._on_update(key) - - def clear(self) -> None: - super().clear() - self._on_update() - - -class _UpdatingCompletable(_UpdatingMixin, _Completable): - - def complete(self) -> None: - super().complete() # type: ignore - self._on_update() - - -class _UpdatingCompletableStringsSet( - _UpdatingCompletable, _CompletableStringsSet): - pass - - -class _UpdatingCompletableStringsOrdered( - _UpdatingCompletable, _CompletableStringsOrdered): - pass - - @dataclass class IrcConnSetup: 'All we need to know to set up a new Client connection.' diff --git a/src/ircplom/client_tui.py b/src/ircplom/client_tui.py index dd6adf3..5e53fdd 100644 --- a/src/ircplom/client_tui.py +++ b/src/ircplom/client_tui.py @@ -5,13 +5,14 @@ from pathlib import Path from tomllib import load as toml_load from typing import Any, Callable, Optional, Sequence # ourselves -from ircplom.tui_base import ( - BaseTui, FormattingString, PromptWidget, TuiEvent, Window, - CMD_SHORTCUTS, LOG_FMT_ATTRS, LOG_FMT_TAG_ALERT) from ircplom.client import ( - AutoAttrMixin, Channel, ChatMessage, Client, ClientQueueMixin, Dict, - DictItem, ImplementationFail, IrcConnSetup, NewClientEvent, NickUserHost, - SendFail, ServerCapability, SharedClientDbFields, TargetUserOffline, User) + Channel, ChatMessage, Client, ClientQueueMixin, ImplementationFail, + IrcConnSetup, NewClientEvent, NickUserHost, SendFail, + ServerCapability, SharedClientDbFields, TargetUserOffline, User) +from ircplom.db_primitives import AutoAttrMixin, Dict, DictItem +from ircplom.tui_base import ( + BaseTui, FormattingString, PromptWidget, TuiEvent, Window, + CMD_SHORTCUTS, LOG_FMT_ATTRS, LOG_FMT_TAG_ALERT) from ircplom.irc_conn import IrcMessage CMD_SHORTCUTS['disconnect'] = 'window.disconnect' diff --git a/src/ircplom/db_primitives.py b/src/ircplom/db_primitives.py new file mode 100644 index 0000000..534ac67 --- /dev/null +++ b/src/ircplom/db_primitives.py @@ -0,0 +1,239 @@ +'Data structuring primitives used by Client databases.' +from abc import ABC, abstractmethod +from typing import (Any, Callable, Collection, Generic, Iterable, Iterator, + Optional, Set, TypeVar) + + +class AutoAttrMixin: + 'Ensures attribute as defined by annotations along MRO' + + @classmethod + def _deep_annotations(cls) -> dict[str, type]: + types: dict[str, type] = {} + for c in cls.__mro__: + if hasattr(c, '__annotations__'): + types = c.__annotations__ | types + return {k: v for k, v in types.items() if k[0] != '_'} + + def __getattribute__(self, key: str): + if key[0] != '_' and (cls := self._deep_annotations().get(key, None)): + try: + return super().__getattribute__(key) + except AttributeError: + setattr(self, key, self._make_attr(cls, key)) + return super().__getattribute__(key) + + +class _Clearable(ABC): + + @abstractmethod + def clear(self) -> None: + 'Zero internal knowledge.' + + +DictItem = TypeVar('DictItem') + + +class Dict(_Clearable, Generic[DictItem]): + 'Customized dict replacement.' + + def __init__(self, **kwargs) -> None: + self._dict: dict[str, DictItem] = {} + super().__init__(**kwargs) + + def keys(self) -> tuple[str, ...]: + 'Keys of item registrations.' + return tuple(self._dict.keys()) + + def __getitem__(self, key: str) -> DictItem: + return self._dict[key] + + def clear(self) -> None: + self._dict.clear() + + @property + def _item_cls(self): + orig_cls = (self.__orig_class__ if hasattr(self, '__orig_class__') + else self.__orig_bases__[0]) + return orig_cls.__args__[0] + + +class _Dict(Dict[DictItem]): + + def __setitem__(self, key: str, val: DictItem) -> None: + assert isinstance(val, self._item_cls) + self._dict[key] = val + + def __delitem__(self, key: str) -> None: + del self._dict[key] + + def values(self) -> tuple[DictItem, ...]: + 'Items registered.' + return tuple(self._dict.values()) + + @staticmethod + def key_val_from_eq_str(eq_str: str) -> tuple[str, str]: + 'Split eq_str by "=", and if none, into eq_str and "".' + toks = eq_str.split('=', maxsplit=1) + return toks[0], '' if len(toks) == 1 else toks[1] + + +class _Completable(ABC): + completed: Any + + @abstractmethod + def complete(self) -> None: + 'Set .completed to "complete" value if possible of current state.' + + +class _CompletableStringsCollection(_Completable, Collection): + _collected: Collection[str] + completed: Optional[Collection[str]] = None + + @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 + + +class _CompletableStringsSet(_CompletableStringsCollection, Set): + _collected: set[str] + completed: Optional[set[str]] = None + + def __init__(self) -> None: + self._collected = set() + + def _copy_collected(self) -> set[str]: + return set(self._collected) + + def _on_set(self, m_name: str, item: str, on_complete: bool, exists: bool + ) -> None: + method = getattr(self._collected, m_name) + if on_complete: + 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 (item in self._collected) == exists + method(item) + + def completable_add(self, item: str, on_complete: bool) -> None: + 'Put item into collection.' + self._on_set('add', item, on_complete, False) + + def completable_remove(self, item: str, on_complete: bool) -> None: + 'Remove item from collection.' + self._on_set('remove', item, on_complete, True) + + def intersection(self, *others: Iterable[str]) -> set[str]: + 'Compare self to other set(s).' + assert self.completed is not None + result = self.completed + for other in others: + assert isinstance(other, set) + result = result.intersection(other) + return result + + +class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection): + _collected: tuple[str, ...] = tuple() + completed: Optional[tuple[str, ...]] = tuple() + + def _copy_collected(self) -> tuple[str, ...]: + return tuple(self._collected) + + def append(self, value: str, complete=False) -> None: + 'Append value to list.' + self._collected += (value,) + if complete: + self.complete() + + def clear(self) -> None: + self._collected = tuple() + self.complete() + + +class _UpdatingMixin: + + def __init__(self, on_update: Callable, **kwargs) -> None: + super().__init__(**kwargs) + self._on_update = on_update + + +class _UpdatingAttrsMixin(_UpdatingMixin, AutoAttrMixin): + + def _make_attr(self, cls: Callable, key: str): + return cls(on_update=lambda *steps: self._on_update(key, *steps)) + + def __setattr__(self, key: str, value) -> None: + super().__setattr__(key, value) + 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: + kw = {} | self._create_if_none + if _UpdatingMixin in self._item_cls.__mro__: + kw |= {'on_update': + lambda *steps: self._on_update(key, *steps)} + self[key] = self._item_cls(**kw) + return super().__getitem__(key) + + def __setitem__(self, key: str, val: DictItem) -> None: + super().__setitem__(key, val) + self._on_update(key) + + def __delitem__(self, key: str) -> None: + super().__delitem__(key) + self._on_update(key) + + def clear(self) -> None: + super().clear() + self._on_update() + + +class _UpdatingCompletable(_UpdatingMixin, _Completable): + + def complete(self) -> None: + super().complete() # type: ignore + self._on_update() + + +class _UpdatingCompletableStringsSet( + _UpdatingCompletable, _CompletableStringsSet): + pass + + +class _UpdatingCompletableStringsOrdered( + _UpdatingCompletable, _CompletableStringsOrdered): + pass