_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:
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]