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 (
'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.'
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'
--- /dev/null
+'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
projects / ircplom / commitdiff