Source code for hbutils.collection.recover

from typing import TypeVar, Type, List, Callable, Optional, Any, Dict

__all__ = [
    'BaseRecovery',
    'DictRecovery', 'ListRecovery', 'TupleRecovery',
    'NullRecovery', 'GenericObjectRecovery',
    'register_recovery', 'get_recovery_func',
]

_OriginType = TypeVar('_OriginType')


[docs]class BaseRecovery: __rtype__ = object
[docs] def __init__(self, origin: _OriginType): """ Constructor of :class:`BaseRecovery`. :param origin: Origin object to be recovered. """ self.origin = origin
[docs] def _recover(self): """ Implementation for recovery. """ raise NotImplementedError # pragma: no cover
[docs] def recover(self) -> _OriginType: """ Recover the given object. :return: Recovered object. """ self._recover() return self.origin
[docs] @classmethod def _recover_child(cls, child): """ Get recovered child-level object. :param child: Child object, should be a :class:`BaseRecovery` or native object. :return: Recovered child-level object. """ if isinstance(child, BaseRecovery): return child.recover() else: return child
[docs] @classmethod def from_origin(cls, origin: _OriginType, recursive: bool = True) -> 'BaseRecovery': """ Create a recovery object by the given original object. :param origin: Original object to recover. :param recursive: Recursive or not. Default is ``True`` which means the child-level object \ contained in ``origin`` will be recovered as well. :return: Recovery object. """ raise NotImplementedError # pragma: no cover
[docs] @classmethod def _create_child(cls, child, recursive: bool = True): """ Create child-level object for storage usage. :param child: Original child-level object. :param recursive: Recursive or not. Default is ``True`` which means the child-level object \ contained in ``origin`` will be recovered as well. :return: Object for storage. """ if recursive: clazz = _get_recovery_class(child) if clazz is not None: return clazz.from_origin(child, recursive) return child
_REC_CLASSES: Optional[List[Type[BaseRecovery]]] = None
[docs]def register_recovery(cls: Type[BaseRecovery]): """ Overview: Register recovery class. :param cls: Recovery class. .. note:: This API is used for customize recovery for other classes. \ For more details, you may take a look at the source code of :class:`BaseRecovery`. """ _REC_CLASSES.append(cls)
_DictType = TypeVar('_DictType', bound=dict)
[docs]class DictRecovery(BaseRecovery): __rtype__ = dict
[docs] def __init__(self, origin: _DictType, mp: Dict): """ Constructor of :class:`DictRecovery`. :param origin: Origin object to be recovered. """ BaseRecovery.__init__(self, origin) self.mapping = mp
[docs] def _recover(self): target = { key: self._recover_child(value) for key, value in self.mapping.items() } keys = set(self.origin.keys()) | set(target.keys()) for key in keys: if key not in target: del self.origin[key] else: self.origin[key] = target[key]
[docs] @classmethod def from_origin(cls, origin: _DictType, recursive: bool = True) -> 'DictRecovery': return cls(origin, {key: cls._create_child(value, recursive) for key, value in origin.items()})
_TupleType = TypeVar('_TupleType', bound=tuple)
[docs]class TupleRecovery(BaseRecovery): __rtype__ = tuple
[docs] def __init__(self, origin: _TupleType, items: List[Any]): """ Constructor of :class:`TupleRecovery`. :param origin: Origin object to be recovered. """ BaseRecovery.__init__(self, origin) self.items = items
[docs] def _recover(self): for item in self.items: self._recover_child(item)
[docs] @classmethod def from_origin(cls, origin: _TupleType, recursive: bool = True) -> 'TupleRecovery': return cls(origin, [cls._create_child(item, recursive) for item in origin])
_ListType = TypeVar('_ListType', bound=list)
[docs]class ListRecovery(BaseRecovery): __rtype__ = list
[docs] def __init__(self, origin: _ListType, items: List): """ Constructor of :class:`ListRecovery`. :param origin: Origin object to be recovered. """ BaseRecovery.__init__(self, origin) self.items = items
[docs] def _recover(self): target = [ self._recover_child(item) for item in self.items ] self.origin[:] = target
[docs] @classmethod def from_origin(cls, origin: _ListType, recursive: bool = True) -> 'ListRecovery': return cls(origin, [cls._create_child(item, recursive) for item in origin])
[docs]class NullRecovery(BaseRecovery): """ Overview: Empty recovery class for builtin immutable types. """ __rtype__ = (int, float, str, bool, bytes, complex, range, slice)
[docs] def _recover(self): """ Just do nothing. """ pass
[docs] @classmethod def from_origin(cls, origin: _OriginType, recursive: bool = True) -> 'NullRecovery': """ Just do nothing. """ return cls(origin)
[docs]class GenericObjectRecovery(BaseRecovery): """ Overview: Recovery class for generic objects. The ``__dict__`` will be recovered. .. note:: If what you need to recover is not only ``__dict__``, may be you need to custom \ recovery class by inheriting :class:`BaseRecovery` class, and register it by \ :func:`register_recovery` function. """ __rtype__ = object
[docs] def __init__(self, origin: _OriginType, dict_: Optional['DictRecovery']): """ Constructor of :class:`GenericObjectRecovery`. :param origin: Original object to recover. :param dict_: Recovery object of ``__dict__``, ``None`` when ``origin`` do not have ``__dict__`` """ BaseRecovery.__init__(self, origin) self.dict_ = dict_
[docs] def _recover(self): """ Recover the ``__dict__``. """ if self.dict_ is not None: self.dict_.recover()
[docs] @classmethod def from_origin(cls, origin: _OriginType, recursive: bool = True) -> 'BaseRecovery': """ Create recovery object. """ dict_ = DictRecovery.from_origin(origin.__dict__, recursive) if hasattr(origin, '__dict__') else None return cls(origin, dict_)
if _REC_CLASSES is None: _REC_CLASSES = [] register_recovery(GenericObjectRecovery) register_recovery(NullRecovery) register_recovery(TupleRecovery) register_recovery(DictRecovery) register_recovery(ListRecovery) def _get_recovery_class(origin: _OriginType) -> Optional[Type[BaseRecovery]]: for cls in reversed(_REC_CLASSES): if isinstance(origin, cls.__rtype__): return cls assert False, f'The object cannot be wrapped by recoveries - {origin!r}' # pragma: no cover
[docs]def get_recovery_func(origin: _OriginType, recursive: bool = True) -> Callable[[], _OriginType]: """ Overview: Get recovery function for given object. Dict, list and tuple object are natively supported. :param origin: Original object to recover. :param recursive: Recursive or not. Default is ``True`` which means the child-level object \ contained in ``origin`` will be recovered as well. :return: Recovery function. Examples:: >>> from hbutils.collection import get_recovery_func >>> l = [1, {'a': 1, 'b': 2}, 3, 4, 5] >>> print(id(l), l) 140146367304720 [1, {'a': 1, 'b': 2}, 3, 4, 5] >>> f = get_recovery_func(l) >>> l[3] = 1 >>> l.pop() >>> l.append('sdklfj') >>> l.append('sdkfhjksd') >>> l[1]['c'] = 2 >>> l[1]['a'] = 100 >>> print(id(l), l) 140146367304720 [1, {'a': 100, 'b': 2, 'c': 2}, 3, 1, 'sdklfj', 'sdkfhjksd'] >>> lx = f() >>> print(id(lx), lx) # lx is l 140146367304720 [1, {'a': 1, 'b': 2}, 3, 4, 5] >>> print(id(l), l) # the value is recovered 140146367304720 [1, {'a': 1, 'b': 2}, 3, 4, 5] """ cls = _get_recovery_class(origin) return cls.from_origin(origin, recursive).recover