# Copyright Swiss Data Science Center (SDSC). A partnership between
# École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Slots and Immutable classes."""
import threading
import weakref
from typing import Tuple, cast
[docs]class Slots:
"""An immutable class.
Subclasses are supposed to use __slots__ to define their members.
"""
__slots__ = ("__weakref__",)
__all_slots__: Tuple[str, ...] = tuple()
def __new__(cls, *args, **kwargs):
"""Create and return an empty instance of the class."""
if not cls.__all_slots__:
cls.__all_slots__ = cast(Tuple[str, ...], cls._get_all_slots())
return object.__new__(cls)
def __init__(self, **kwargs):
for key, value in kwargs.items():
object.__setattr__(self, key, value)
[docs] @classmethod
def make_instance(cls, **kwargs):
"""Instantiate from the given parameters."""
return cls(**kwargs)
def __getstate__(self):
return {name: getattr(self, name, None) for name in self.__class__.__all_slots__ if name != "__weakref__"}
def __setstate__(self, state):
new_attributes = set(self.__all_slots__) - set(state)
for name in new_attributes:
if name != "__weakref__":
object.__setattr__(self, name, None)
for name, value in state.items():
object.__setattr__(self, name, value)
@classmethod
def _get_all_slots(cls):
all_slots = set()
for klass in cls.mro():
if not hasattr(klass, "__slots__"):
continue
slots = {klass.__slots__} if isinstance(klass.__slots__, str) else set(klass.__slots__)
all_slots.update(slots)
return tuple(all_slots)
[docs]class Immutable(Slots):
"""An immutable class that its instances can be cached and reused.
Immutable subclasses should only contain immutable members. They must call super().__init__(...) to initialize their
instances. They should not redefine `id` attribute.
NOTE: Immutable objects must not be modified during the whole provenance and not just only during the object's
lifetime. These is because we cache these objects and return a single instance if an object with the same `id`
exists. For example, a DatasetFile object is not immutable across the whole provenance because once it gets removed
its `date_removed` attribute is set which make the object different from a previous version. As a rule of thumb, an
object can be immutable if all of its attributes values appear in its id.
"""
__slots__ = ("id",)
_local = threading.local()
[docs] @classmethod
def make_instance(cls, instance):
"""Return a cached instance if available otherwise create an instance from the given parameters."""
if getattr(cls._local, "cache", None) is None:
cls._local.cache = weakref.WeakValueDictionary()
id = getattr(instance, "id", None)
existing_instance = cls._local.cache.get(id)
if existing_instance:
return existing_instance
if id is not None:
cls._local.cache[id] = instance
return instance
def __setattr__(self, name, value):
if name != "__weakref__":
raise TypeError(f"Cannot modify an immutable class {self} {self.__class__}")
object.__setattr__(self, name, value)
@property
def __name__(self):
return self.__class__.__name__
[docs]class DynamicProxy:
"""A proxy class to allow adding dynamic fields to slots/immutable classes."""
def __init__(self, subject, update=True):
self._subject = subject
self._update = update
def __getattr__(self, name):
return getattr(self._subject, name)
def __getattribute__(self, name: str):
if name == "__class__":
# NOTE: Makes isinstance() checks work with proxies
return object.__getattribute__(self._subject, "__class__")
return object.__getattribute__(self, name)
def __setattr__(self, name, value):
if name == "_subject" or not hasattr(self._subject, name):
super().__setattr__(name, value)
else:
if not self._update:
raise ValueError(f"Cannot set attribute '{name}' on {self._subject}.")
setattr(self._subject, name, value)