From: Christian Heller Date: Sat, 20 Sep 2025 00:08:53 +0000 (+0200) Subject: Minor code re-arrangement for better readability. X-Git-Url: https://plomlompom.com/repos/booking/pick_tasks?a=commitdiff_plain;h=19ab4535a8c03140404a4f83c37bb0220cd8a28f;p=ircplom Minor code re-arrangement for better readability. --- diff --git a/ircplom/client_tui.py b/ircplom/client_tui.py index 99c2288..468216c 100644 --- a/ircplom/client_tui.py +++ b/ircplom/client_tui.py @@ -23,90 +23,6 @@ _LOG_PREFIX_OUT = '>' _LOG_PREFIX_IN = '<' -class _Update: - old_value: Any - results: list[tuple[LogScope, Any]] - - def __init__(self, path: tuple[str, ...], value: Any) -> None: - self.full_path = path - self.rel_path = self.full_path[:] - self.value = value - self.old_value = None - self.force_log = False - self.results = [] - - @property - def key(self) -> str: - 'Name of item or attribute to be processed.' - return self.rel_path[0] - - def decrement_path(self) -> None: - 'Remove first element from .rel_path.' - self.rel_path = self.rel_path[1:] - - -class _UpdatingNode(AutoAttrMixin): - - def _make_attr(self, cls: Callable, key: str): - return cls() - - def recursive_set_and_report_change(self, update: _Update) -> None: - 'Apply update, and, if it makes a difference, add to its .results.' - update.force_log = update.force_log or (not self._is_set(update.key)) - node = self._get(update.key) - if len(update.rel_path) > 1: - update.decrement_path() - node.recursive_set_and_report_change(update) - else: - update.old_value = node - do_report = update.force_log - if update.value is None: - if self._is_set(update.key): - self._unset(update.key) - do_report |= True - elif update.old_value != update.value: - self._set(update.key, update.value) - do_report |= True - if do_report: - update.results += [(LogScope.SERVER, - tuple(sorted(update.value)) - if isinstance(update.value, set) - else update.value)] - - 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() - - def _is_set(self, key: str) -> bool: - return hasattr(self, key) - - -class _UpdatingDict(Dict[DictItem], _UpdatingNode): - - def items(self) -> tuple[tuple[str, DictItem], ...]: - 'Key-value pairs of item registrations.' - return tuple((k, v) for k, v in self._dict.items()) - - def _get(self, key: str): - if key not in self._dict: - self._dict[key] = self._item_cls() - return self._dict[key] - - def _set(self, key: str, value) -> None: - self._dict[key] = value - - def _unset(self, key: str) -> None: - del self._dict[key] - - def _is_set(self, key: str) -> bool: - return key in self._dict - - class _ClientWindow(Window, ClientQueueMixin): def __init__(self, scope: LogScope, log: Callable, **kwargs) -> None: @@ -204,6 +120,90 @@ class _QueryWindow(_ChatWindow): pass +class _Update: + old_value: Any + results: list[tuple[LogScope, Any]] + + def __init__(self, path: tuple[str, ...], value: Any) -> None: + self.full_path = path + self.rel_path = self.full_path[:] + self.value = value + self.old_value = None + self.force_log = False + self.results = [] + + @property + def key(self) -> str: + 'Name of item or attribute to be processed.' + return self.rel_path[0] + + def decrement_path(self) -> None: + 'Remove first element from .rel_path.' + self.rel_path = self.rel_path[1:] + + +class _UpdatingNode(AutoAttrMixin): + + def _make_attr(self, cls: Callable, key: str): + return cls() + + def recursive_set_and_report_change(self, update: _Update) -> None: + 'Apply update, and, if it makes a difference, add to its .results.' + update.force_log = update.force_log or (not self._is_set(update.key)) + node = self._get(update.key) + if len(update.rel_path) > 1: + update.decrement_path() + node.recursive_set_and_report_change(update) + else: + update.old_value = node + do_report = update.force_log + if update.value is None: + if self._is_set(update.key): + self._unset(update.key) + do_report |= True + elif update.old_value != update.value: + self._set(update.key, update.value) + do_report |= True + if do_report: + update.results += [(LogScope.SERVER, + tuple(sorted(update.value)) + if isinstance(update.value, set) + else update.value)] + + 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() + + def _is_set(self, key: str) -> bool: + return hasattr(self, key) + + +class _UpdatingDict(Dict[DictItem], _UpdatingNode): + + def items(self) -> tuple[tuple[str, DictItem], ...]: + 'Key-value pairs of item registrations.' + return tuple((k, v) for k, v in self._dict.items()) + + def _get(self, key: str): + if key not in self._dict: + self._dict[key] = self._item_cls() + return self._dict[key] + + def _set(self, key: str, value) -> None: + self._dict[key] = value + + def _unset(self, key: str) -> None: + del self._dict[key] + + def _is_set(self, key: str) -> bool: + return key in self._dict + + class _UpdatingChannel(_UpdatingNode, Channel): user_ids: set[str] exits: _UpdatingDict[str]