Source code for dantro.mixins.base

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

import contextlib
import logging
import sys
import warnings
from typing import Union

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

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 :py:class:`~dantro.base.BaseDataAttrs` -derived object and store it as an attribute. This relays the checking of the correct attribute format to the actual :py:class:`~dantro.base.BaseDataAttrs`-derived class. For changing the class that is used for the attributes, an overwrite of the ``_ATTRS_CLS`` class variable suffices. """ _attrs = None """The class attribute that the attributes will be stored to""" _ATTRS_CLS = None """The class to use for storing attributes""" @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!" ) 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 :py:func:`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. """ 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. """ __locked = False """Whether the data is regarded as locked. Note name-mangling here.""" @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 :py:meth:`._check_data` method, overwriting the placeholder method in the :py:class:`~dantro.base.BaseDataContainer`, and can be controlled via class variables. .. note:: This is not suitable for checking containers that are added to an object of a :py:class:`~dantro.base.BaseDataGroup`-derived class! """ DATA_EXPECTED_TYPES: tuple = None """Which types to allow. If None, all types are allowed.""" DATA_ALLOW_PROXY: bool = False """Whether to allow *all* proxy types, i.e. classes derived from :py:class:`~dantro.abc.AbstractDataProxy`.""" DATA_UNEXPECTED_ACTION = "warn" """The action to take when an unexpected type was supplied. 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 """ from .._import_tools import resolve_types if self.DATA_EXPECTED_TYPES is None: # All types allowed return # Compile tuple of allowed types, importing those that were supplied # as module strings expected_types = resolve_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 """A name-mangled state flag that determines the state of the object.""" @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:`.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:`._enter_direct_insertion_mode` is called. *After* exiting, :py:meth:`_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