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 (
client_id=self.client_id).kw(e=e)
-class _CompletableTopic(_Completable, Topic):
+class _CompletableTopic(Completable, Topic):
completed: Optional[Topic] = None
def complete(self) -> None:
class _Channel(Channel):
- user_ids: _CompletableStringsSet
+ user_ids: CompletableStringsSet
topic: _CompletableTopic
def __init__(self,
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
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}
channel.remove_user(user, msg)
-class _UpdatingIsupportDict(_UpdatingDict[str]):
+class _UpdatingIsupportDict(UpdatingDict[str]):
def __delitem__(self, key: str) -> None:
if key in ISUPPORT_DEFAULTS:
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:
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 []
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:
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, '')
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)
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()}
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:
del self._dict[key]
-class _Completable(ABC):
+class Completable(ABC):
+ 'Data whose collection can be declared complete/incomplete.'
completed: Any
@abstractmethod
'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
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
return result
-class _CompletableStringsOrdered(_Clearable, _CompletableStringsCollection):
+class _CompletableStringsOrdered(Clearable, _CompletableStringsCollection):
_collected: tuple[str, ...] = tuple()
completed: Optional[tuple[str, ...]] = tuple()
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))
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:
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)
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.'
projects / ircplom / commitdiff