from collections.abc import MutableSequence, MutableMapping, Mapping from typing import ( Any, Optional, Tuple, List, Dict, Union, Sequence, Iterable, TypeVar, Generic, Callable, overload, cast, ItemsView ) from biopro.util.json import JSON, JSON_ARRAY, JSON_OBJECT, JsonSerialize __all__ = ['T', '_truncate', '_get', '_list', '_dict', '_unique', 'ImmutableListNode', 'ListNode', 'ImmutableDictNode', 'DictNode', 'JSON', 'JSON_ARRAY', 'JSON_OBJECT', 'JsonSerialize'] T = TypeVar('T') J = TypeVar('J', bound=JsonSerialize) def _truncate(json: Any, limit=32, symbol='...') -> str: r = str(json) if len(r) > limit: r = r[:limit] + symbol return r def _get(json: JSON_OBJECT, key: str, ty: Union[None, type, Tuple[type, ...]], message: str, nullable=False, default: T = None) -> T: j = json.get(key, None) if j is None: if nullable: return default else: raise RuntimeError(message + ' lost ' + key + ' : ' + _truncate(json)) if ty is not None: if not isinstance(j, ty): dump = _truncate(j) if ty == dict: raise RuntimeError(message + ' not a object : ' + dump) elif ty == list: raise RuntimeError(message + ' not a array : ' + dump) else: raise RuntimeError(message + ' not a ' + str(ty) + ' : ' + dump) return j def _list(json: JSON_ARRAY, e: Callable[[JSON], J]) -> List[J]: return [e(v) for v in json] def _dict(json: JSON_OBJECT, e: Callable[[str, Any], J]) -> Dict[str, J]: return {k: e(k, v) for k, v in json.items()} def _unique(ls: Iterable[T]) -> Iterable[T]: s = set() for it in ls: if it in s: continue else: s.add(it) yield it class ImmutableListNode(Sequence, JsonSerialize, Generic[J]): """User defined list whose contain element should a json type or a derived class of :class:`JsonSerialize`. This list is immutable, which mean user can not modify the element in this list (do not include the content of the element) through the public API. """ def __init__(self, init_list: Iterable[J]): """ :param init_list: initial list """ self._list = list(init_list) '''object''' def __repr__(self): return repr(self._list) def __str__(self): return str(self._list) '''Sequence''' def __len__(self): return len(self._list) def __contains__(self, item) -> bool: return item in self._list @overload def __getitem__(self, idx: int) -> J: pass @overload def __getitem__(self, idx: slice) -> Sequence[J]: pass def __getitem__(self, idx): return self._list[idx] def __iter__(self) -> Iterable[J]: return iter(self._list) def copy(self): """unsupported""" raise RuntimeError('unsupported') def count(self, item: J): return self._list.count(item) def index(self, item: J, start=0, stop=None): stop = stop if stop is not None else len(self._list) return self._list.index(item, start=start, stop=stop) '''JsonSerialize''' def as_json(self): return [JsonSerialize.to_json(v) for v in self._list] class ListNode(ImmutableListNode[J], MutableSequence): __slots__ = () def __init__(self, init_list: List[J] = None): """ :param init_list: initial list """ super().__init__(init_list if init_list is not None else []) '''MutableSequence''' @overload def __setitem__(self, idx: int, value: J): pass @overload def __setitem__(self, idx: slice, value: List[J]): pass def __setitem__(self, idx, value): self._list[idx] = value def __delitem__(self, idx: Union[int, slice]): del self._list[idx] def append(self, item: J): self._list.append(item) def insert(self, idx: int, item: J): self._list.insert(idx, item) def pop(self, idx: int = -1): return self._list.pop(idx) def remove(self, item: J): self._list.remove(item) def clear(self): self._list.clear() def reverse(self): self._list.reverse() def sort(self, key: Optional[Callable[[J], Any]] = None, reverse=False): self._list.sort(key=key, reverse=reverse) def extend(self, other: List[J]): self._list.extend(other) '''operator overrode''' def __add__(self, other: Union[J, List[J]]) -> 'ListNode[J]': if isinstance(other, ImmutableListNode): self._list.extend(other._list) elif isinstance(other, list): self._list.extend(other) else: self._list.append(cast(J, other)) return self class ImmutableDictNode(Mapping, JsonSerialize, Generic[J]): """User defined dictionary whose key should be a :class:`str` and the value should a json type or a derived class of :class:`JsonSerialize`. This dict is immutable, which mean user can not modify the element in this dict (do not include the content of the element) through the public API. """ def __init__(self, init_dict: Dict[str, J]): """ :param init_dict: initial dictionary """ self._dict = dict(init_dict) '''object''' def __repr__(self): return repr(self._dict) def __str__(self): return str(self._dict) '''Mapping''' def __len__(self): return len(self._dict) def __contains__(self, key: Any) -> bool: return key in self._dict def __getitem__(self, key: str) -> J: """get the item with *key*. If *key* not found in this dict, program will try to call method ``__missing__(key)`` if defined. otherwise, raise :class:`KeyError`. :param key: :return: """ if key in self._dict: return self._dict[key] elif hasattr(self.__class__, '__missing__'): # noinspection PyUnresolvedReferences return self.__class__.__missing__(self, key) else: raise KeyError(key) def __iter__(self) -> Iterable[str]: return iter(self._dict.keys()) def copy(self): """unsupported""" raise RuntimeError('unsupported') '''dict''' def keys(self) -> Iterable[str]: return self._dict.keys() def values(self) -> Iterable[J]: return self._dict.values() def items(self) -> ItemsView[str, J]: return self._dict.items() '''JsonSerialize''' def as_json(self) -> Dict[str, Any]: return {k: JsonSerialize.to_json(v) for k, v in self._dict.items()} class DictNode(ImmutableDictNode[J], MutableMapping): __slots__ = () def __init__(self, init_dict: Dict[str, J] = None): """ :param init_dict: initial dictionary """ super().__init__(init_dict if init_dict is not None else {}) '''MutableMapping''' def __setitem__(self, key: str, value: J): self._dict[key] = value def __delitem__(self, key: str): del self._dict[key]