Source code for mincepy.types

# -*- coding: utf-8 -*-
from abc import ABCMeta, abstractmethod
import datetime
from typing import Type, List
import uuid

try:  # Python3
    from hashlib import blake2b
except ImportError:  # Python < 3.6
    from pyblake2 import blake2b

from . import depositors
from . import expr
from . import fields
from . import migrations  # pylint: disable=unused-import
from . import saving
from . import tracking

__all__ = "Savable", "Comparable", "Object", "SavableObject", "PRIMITIVE_TYPES"

# The primitives that all archive types must support

def is_primitive(obj):
    return obj.__class__ in PRIMITIVE_TYPES

[docs]class Savable(fields.WithFields, expr.FilterLike): """Interface for an object that can save and load its instance state""" TYPE_ID = None LATEST_MIGRATION: "migrations.ObjectMigration" = None def __init__(self, *args, **kwargs): assert ( self.TYPE_ID is not None ), "Must set the TYPE_ID for an object to be savable" super().__init__(*args, **kwargs) @classmethod def __expr__(cls): """This method gives savables the ability to be used as an expression""" return expr.Comparison("type_id", expr.Eq(cls.TYPE_ID)) @classmethod def __query_expr__(cls) -> dict: """This method gives savables the ability to be used in query filter expressions""" return cls.__expr__().__query_expr__()
[docs] def save_instance_state( self, saver: depositors.Saver ): # pylint: disable=unused-argument """Save the instance state of an object, should return a saved instance""" return saving.save_instance_state(self)
[docs] def load_instance_state( self, saved_state, loader: depositors.Loader ): # pylint: disable=unused-argument """Take the given object and load the instance state into it""" saving.load_instance_state(self, saved_state)
[docs]class Comparable(metaclass=ABCMeta): """Interface for an object that can be compared and hashed""" @abstractmethod def __eq__(self, other) -> bool: """Determine if two objects are equal"""
[docs] @abstractmethod def yield_hashables(self, hasher): """Produce a hash representing the value"""
[docs]class Object(Comparable, metaclass=ABCMeta): """A simple object that is comparable"""
[docs]class SavableObject(Object, Savable, metaclass=ABCMeta): """A class that is both savable and comparable""" _historian = None @classmethod def init_field(cls, field: fields.Field, attr_name: str): super().init_field(field, attr_name) field.set_query_context(expr.Comparison("type_id", expr.Eq(cls.TYPE_ID))) field.path_prefix = "state" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) tracking.obj_created(self) def __eq__(self, other) -> bool: """Determine if two objects are equal""" if not isinstance(other, type(self)): return False return saving.save_instance_state(self) == saving.save_instance_state(other)
[docs] def yield_hashables(self, hasher): """Produce a hash representing the object""" yield from hasher.yield_hashables(saving.save_instance_state(self))
class Equator: def __init__(self, equators=tuple()): self._equators = list(equators) def do_hash(*args): hasher = blake2b(digest_size=32) for arg in args: hasher.update(arg) return hasher.hexdigest() self._hasher = do_hash def add_equator(self, equator): self._equators.append(equator) def remove_equator(self, equator): self._equators.reverse() try: self._equators.remove(equator) except ValueError as exc: raise ValueError(f"Unknown equator '{equator}'") from exc finally: self._equators.reverse() def get_equator(self, obj): # Iterate in reversed order i.e. the latest added should be used preferentially for equator in reversed(self._equators): if isinstance(obj, equator.TYPE): return equator raise TypeError( f"Don't know how to compare '{type(obj)}' types, no type equator set" ) def yield_hashables(self, obj): try: equator = self.get_equator(obj) except TypeError: # Try the objects' method try: yield from obj.yield_hashables(self) except AttributeError: raise TypeError( f"No helper registered and no yield_hashables method on '{type(obj).__name__}'" ) from None else: yield from equator.yield_hashables(obj, self) def hash(self, obj): return self._hasher(*self.yield_hashables(obj)) def eq(self, obj1, obj2) -> bool: # pylint: disable=invalid-name if not type(obj1) == type(obj2): # pylint: disable=unidiomatic-typecheck return False try: equator = self.get_equator(obj1) except TypeError: # Fallback to python eq return obj1 == obj2 else: return equator.eq(obj1, obj2) def float_to_str(self, value, sig=14): """ Convert float to text string for computing hash. Preserve up to N significant number given by sig. :param value: the float value to convert :param sig: choose how many digits after the comma should be output """ fmt = f"{{:.{sig}g}}" return fmt.format(value) def is_savable_type(obj_type: Type) -> bool: return issubclass(obj_type, SavableObject) and obj_type.TYPE_ID is not None def savable_mro(obj_type: Type[SavableObject]) -> List[Type[SavableObject]]: """Given a SavableObject type this will give the mro of the savable types in the hierarchy""" mro = obj_type.mro() return list(filter(is_savable_type, mro))