Source code for dantro.mixins.base

"""This sub-module implements the basic mixin classes that are required
in the dantro.base module"""

import contextlib
import logging
import sys
import warnings

from ..abc import PATH_JOIN_CHAR, AbstractDataProxy
from ..exceptions import UnexpectedTypeWarning

# Local constants
log = logging.getLogger(__name__)


# -----------------------------------------------------------------------------


[docs]class AttrsMixin: """This Mixin class supplies the `attrs` property getter and setter and the private `_attrs` attribute. Hereby, the setter function will initialize a BaseDataAttrs-derived object and store it as an attribute. This relays the checking of the correct attribute format to the actual BaseDataAttrs-derived class. For changing the class that is used for the attributes, an overwrite of the _ATTRS_CLS class variable suffices. """ # The class attribute that the attributes will be stored to _attrs = None # Define the class to use for storing attributes _ATTRS_CLS = None @property def attrs(self): """The container attributes.""" return self._attrs @attrs.setter def attrs(self, new_attrs): """Setter method for the container `attrs` attribute.""" if self._ATTRS_CLS is None: raise ValueError( "Need to declare the class variable _ATTRS_CLS " "in order to use the AttrsMixin!" ) # Perform the initialisation self._attrs = self._ATTRS_CLS(name="attrs", attrs=new_attrs)
[docs]class SizeOfMixin: """Provides the __sizeof__ magic method and attempts to take into account the size of the attributes. """
[docs] def __sizeof__(self) -> int: """Returns the size of the data (in bytes) stored in this container's data and its attributes. Note that this value is approximate. It is computed by calling the ``sys.getsizeof`` function on the data, the attributes, the name and some caching attributes that each dantro data tree class contains. Importantly, this is not a recursive algorithm. Also, derived classes might implement further attributes that are not taken into account either. To be more precise in a subclass, create a specific __sizeof__ method and invoke this parent method additionally. For more information, see the documentation of ``sys.getsizeof``: https://docs.python.org/3/library/sys.html#sys.getsizeof """ nbytes = sys.getsizeof(self._data) nbytes += sys.getsizeof(self._attrs) nbytes += sys.getsizeof(self._name) return nbytes
[docs]class LockDataMixin: """This Mixin class provides a flag for marking the data of a group or container as locked. """ # Whether the data is regarded as locked. Note name-mangling here. __locked = False @property def locked(self) -> bool: """Whether this object is locked""" return self.__locked
[docs] def lock(self): """Locks the data of this object""" self.__locked = True self._lock_hook()
[docs] def unlock(self): """Unlocks the data of this object""" self.__locked = False self._unlock_hook()
[docs] def raise_if_locked(self, *, prefix: str = None): """Raises an exception if this object is locked; does nothing otherwise""" if self.locked: raise RuntimeError( "{}Cannot modify {} because it was already " "marked locked." "".format(prefix + " " if prefix else "", self.logstr) )
[docs] def _lock_hook(self): """Invoked upon locking.""" pass
[docs] def _unlock_hook(self): """Invoked upon unlocking.""" pass
[docs]class BasicComparisonMixin: """Provides a (very basic) ``__eq__`` method to compare equality."""
[docs] def __eq__(self, other) -> bool: """Evaluates equality by making the following comparisons: identity, strict type equality, and finally: equality of the ``_data`` and ``_attrs`` attributes, i.e. the *private* attribute. This ensures that comparison does not trigger any downstream effects like resolution of proxies. If types do not match exactly, ``NotImplemented`` is returned, thus referring the comparison to the other side of the ``==``. """ if other is self: return True if type(other) is not type(self): return NotImplemented return self._data == other._data and self._attrs == other._attrs
[docs]class CollectionMixin: """This Mixin class implements the methods needed for being a Collection. It relays all calls forward to the data attribute. """
[docs] def __contains__(self, key) -> bool: """Whether the given key is contained in the items.""" return bool(key in self.data)
[docs] def __len__(self) -> int: """The number of items.""" return len(self.data)
[docs] def __iter__(self): """Iterates over the items.""" return iter(self.data)
[docs]class ItemAccessMixin: """This Mixin class implements the methods needed for getting, setting, and deleting items. It relays all calls forward to the data attribute, but if given a list (passed down from above), it extracts it """
[docs] def __getitem__(self, key): """Returns an item.""" return self.data[self._item_access_convert_list_key(key)]
[docs] def __setitem__(self, key, val): """Sets an item.""" self.data[self._item_access_convert_list_key(key)] = val
[docs] def __delitem__(self, key): """Deletes an item""" del self.data[self._item_access_convert_list_key(key)]
[docs] def _item_access_convert_list_key(self, key): """If given something that is not a list, just return that key""" if isinstance(key, list): if len(key) > 1: return tuple(key) return key[0] return key
[docs]class MappingAccessMixin(ItemAccessMixin, CollectionMixin): """Supplies all methods that are needed for Mapping access. All calls are relayed to the data attribute. """
[docs] def keys(self): """Returns an iterator over the data's keys.""" return self.data.keys()
[docs] def values(self): """Returns an iterator over the data's values.""" return self.data.values()
[docs] def items(self): """Returns an iterator over data's (key, value) tuples""" return self.data.items()
[docs] def get(self, key, default=None): """Return the value at ``key``, or ``default`` if ``key`` is not available. """ return self.data.get(key, default)
[docs]class CheckDataMixin: """This mixin class extends a BaseDataContainer-derived class to check the provided data before storing it in the container. It implements a general _check_data method, overwriting the placeholder method in the BaseDataContainer, and can be controlled via class variables. .. note:: This is not suitable for checking containers that are added to an object of a BaseDataGroup-derived class! Attributes: DATA_ALLOW_PROXY (bool): Whether to allow _all_ proxy types, i.e. classes derived from AbstractDataProxy DATA_EXPECTED_TYPES (tuple, None): Which types to allow. If None, all types are allowed. DATA_UNEXPECTED_ACTION (str): The action to take when an unexpected type was supplied. Can be: raise, warn, ignore """ # Specify expected data types for this container class DATA_EXPECTED_TYPES = None # as tuple or None (allow all) DATA_ALLOW_PROXY = False # to check for AbstractDataProxy DATA_UNEXPECTED_ACTION = "warn" # Can be: raise, warn, ignore
[docs] def _check_data(self, data) -> None: """A general method to check the received data for its type Args: data: The data to check Raises: TypeError: If the type was unexpected and the action was 'raise' ValueError: Illegal value for DATA_UNEXPECTED_ACTION class variable Returns: None """ if self.DATA_EXPECTED_TYPES is None: # All types allowed return # Compile tuple of allowed types expected_types = self.DATA_EXPECTED_TYPES if self.DATA_ALLOW_PROXY: expected_types += (AbstractDataProxy,) # Check for expected types if isinstance(data, expected_types): return # else: was not of the expected type # Create a base message msg = ( f"Unexpected type {type(data)} for data passed to {self.logstr}! " f"Expected types are: {expected_types}." ) # Handle according to the specified action if self.DATA_UNEXPECTED_ACTION == "raise": raise TypeError(msg) elif self.DATA_UNEXPECTED_ACTION == "warn": warnings.warn( f"{msg}\nInitialization will work, but be informed " "that there might be errors at runtime.", UnexpectedTypeWarning, ) elif self.DATA_UNEXPECTED_ACTION == "ignore": log.debug(msg + " Ignoring ...") else: raise ValueError( f"Illegal value '{self.DATA_UNEXPECTED_ACTION}' for class " f"variable DATA_UNEXPECTED_ACTION of {self.classname}. " "Allowed values are: raise, warn, ignore" )
[docs]class DirectInsertionModeMixin: """A mixin class that provides a context manager, within which insertion into the mixed-in class (think: group or container) can happen more directly. This is useful in cases where more assumptions can be made about the to-be-inserted data, thus allowing to make fewer checks during insertion (think: duplicates, key order, etc.). .. note:: This direct insertion mode is not (yet) part of the public interface, as it has to be evaluated how robust and error-prone it is. """ __in_direct_insertion_mode = False @property def with_direct_insertion(self) -> bool: """Whether the class this mixin is mixed into is currently in direct insertion mode. """ return self.__in_direct_insertion_mode
[docs] @contextlib.contextmanager def _direct_insertion_mode(self, *, enabled: bool = True): """A context manager that brings the class this mixin is used in into direct insertion mode. While in that mode, the :py:meth:`~dantro.mixins.base.DirectInsertionModeMixin.with_direct_insertion` property will return true. This context manager additionally invokes two callback functions, which can be specialized to perform certain operations when entering or exiting direct insertion mode: *Before* entering, :py:meth:`~dantro.mixins.base.DirectInsertionModeMixin._enter_direct_insertion_mode` is called. *After* exiting, :py:meth:`~dantro.mixins.base.DirectInsertionModeMixin._exit_direct_insertion_mode` is called. Args: enabled (bool, optional): whether to actually use direct insertion mode. If False, will yield directly without setting the toggle. This is equivalent to a null-context. """ if not enabled: self.__in_direct_insertion_mode = False yield return log.trace( "Entering direct insertion mode of %s @ %s ...", self.logstr, self.path, ) # Perform the entering callback try: self._enter_direct_insertion_mode() except Exception as exc: raise RuntimeError( "Error in callback while entering direct insertion mode of " f"{self.logstr} @ {self.name}! {type(exc).__name__}: {exc}" ) from exc # Now inside direct insertion mode self.__in_direct_insertion_mode = True try: # Yield control to the with-context now yield finally: # Will end up here if there was an exception within the context. log.trace( "Exiting direct insertion mode of %s @ %s ...", self.logstr, self.path, ) self.__in_direct_insertion_mode = False # NOTE Important to NOT have a return here or handle any other # error, otherwise exceptions from the context are discarded. # Perform the exiting callback try: self._exit_direct_insertion_mode() except Exception as exc: raise RuntimeError( "Error in callback while exiting direct insertion mode of " f"{self.logstr} @ {self.name}! {type(exc).__name__}: {exc}" ) from exc
[docs] def _enter_direct_insertion_mode(self): """Called after entering direct insertion mode; can be overwritten to attach additional behaviour. """ pass
[docs] def _exit_direct_insertion_mode(self): """Called before exiting direct insertion mode; can be overwritten to attach additional behaviour. """ pass