"""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
