Source code for properties.extras.uid

"""Classes for dealing with HasProperties instances with unique IDs"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import uuid

from six import string_types, text_type

from .. import base, basic, handlers, utils


[docs]class HasUID(base.HasProperties): """HasUID is a HasProperties class that includes unique ID Adding a UID to HasProperties allows serialization of more complex structures, including recursive self-references. They are serialized to a flat dictionary of UID/HasUID key/value pairs. """ _REGISTRY = dict() _INSTANCES = dict() uid = basic.String( 'Unique identifier', default=lambda: text_type(uuid.uuid4()), ) def __init__(self, **kwargs): super(HasUID, self).__init__(**kwargs) self._INSTANCES[self.uid] = self @handlers.validator('uid') def _ensure_unique(self, change): if self.uid == change['value']: pass elif change['value'] in self._INSTANCES: raise utils.ValidationError( message='UID already used: {}'.format(change['value']), reason='invalid', prop=change['name'], instance=self, ) return True @handlers.observer('uid') def _update_instances(self, change): self._INSTANCES.update({change['value']: self})
[docs] @classmethod def validate_uid(cls, uid): #pylint: disable=unused-argument """Assert if a given UID is valid This is used by Pointer properties to validate a UID without necessarily loading the corresponding instance. """ return True
[docs] @classmethod def load(cls, uid): """Load an instance given a UID This is used by Pointer properties to retrieve instances from UIDs. """ return cls._INSTANCES.get(uid)
[docs] def serialize(self, include_class=True, save_dynamic=False, **kwargs): """Serialize nested HasUID instances to a flat dictionary **Parameters**: * **include_class** - If True (the default), the name of the class will also be saved to the serialized dictionary under key :code:`'__class__'` * **save_dynamic** - If True, dynamic properties are written to the serialized dict (default: False). * You may also specify a **registry** - This is the flat dictionary where UID/HasUID pairs are stored. By default, no registry need be provided; a new dictionary will be created. * Any other keyword arguments will be passed through to the Property serializers. """ registry = kwargs.pop('registry', None) if registry is None: registry = dict() if not registry: root = True registry.update({'__root__': self.uid}) else: root = False key = self.uid if key not in registry: registry.update({key: None}) registry.update({key: super(HasUID, self).serialize( registry=registry, include_class=include_class, save_dynamic=save_dynamic, **kwargs )}) if root: return registry return key
[docs] @classmethod def deserialize(cls, value, trusted=False, strict=False, assert_valid=False, **kwargs): """Deserialize nested HasUID instance from flat pointer dictionary **Parameters** * **value** - Flat pointer dictionary produced by :code:`serialize` with UID/HasUID key/value pairs. It also includes a :code:`__root__` key to specify the root HasUID instance. * **trusted** - If True (and if the input dictionaries have :code:`'__class__'` keyword and this class is in the registry), the new **HasProperties** class will come from the dictionary. If False (the default), only the **HasProperties** class this method is called on will be constructed. * **strict** - Requires :code:`'__class__'`, if present on the input dictionary, to match the deserialized instance's class. Also disallows unused properties in the input dictionary. Default is False. * **assert_valid** - Require deserialized instance to be valid. Default is False. * You may also specify an alternative **root** - This allows a different HasUID root instance to be specified. It overrides :code:`__root__` in the input dictionary. * Any other keyword arguments will be passed through to the Property deserializers. .. note:: HasUID instances are constructed with no input arguments (ie :code:`cls()` is called). This means deserialization will fail if the init method has been overridden to require input parameters. """ registry = kwargs.pop('registry', None) if registry is None: if not isinstance(value, dict): raise ValueError('HasUID must deserialize from dictionary') registry = value.copy() uid = kwargs.get('root', registry.get('__root__')) else: uid = value if uid in cls._INSTANCES and uid not in registry: return cls._INSTANCES[uid] if uid in cls._INSTANCES: raise ValueError('UID already used: {}'.format(uid)) if uid not in registry: raise ValueError('Invalid UID: {}'.format(uid)) value = registry[uid] if not isinstance(value, HasUID): try: input_class = value.get('__class__') except AttributeError: input_class = None new_cls = cls._deserialize_class(input_class, trusted, strict) new_inst = new_cls() registry.update({uid: new_inst}) super(HasUID, cls).deserialize( value=value, trusted=trusted, strict=strict, registry=registry, _instance=new_inst, **kwargs ) cls._INSTANCES[uid] = registry[uid] return registry[uid]
[docs]class Pointer(base.Instance): """Property for HasUID instances where string UID pointer may be used **Available keywords** (in addition to those inherited from :ref:`Instance <instance>`): * **load** - Attempt to load instances from UID on validation If True, when the Pointer property is assigned a valid UID, it will then attempt to call :code:`self.instance_class.load(uid)` If this method is defined, it must return a valid instance which will replace the UID as the Pointer value. If this method is not defined or if it returns None, the Pointer property maintains the UID value. Default is False, meaning there is no attempt to load the instance. * **uid_prop** - Property or attribute name of the UID property on instance_class. The default is 'uid'. """ class_info = 'an instance or uid of an instance' @property def load(self): """Attempt to load instances from UID on validation If True, when the Pointer property is assigned a valid UID, it will then attempt to call :code:`self.instance_class.load(uid)` If this method is defined, it must return a valid instance which will replace the UID as the Pointer value. If this method is not defined or if it returns None, the Pointer property maintains the UID value. Default is False, meaning there is no attempt to load the instance. """ return getattr(self, '_load', False) @load.setter def load(self, value): self._load = bool(value) @property def uid_prop(self): """Property or attribute name of the UID property on instance_class The default is 'uid' """ return getattr(self, '_uid_prop', 'uid') @uid_prop.setter def uid_prop(self, value): self._uid_prop = text_type(value) @property def info(self): info = '{} or a valid {} property of that class'.format( super(Pointer, self).info, self.uid_prop ) return info def validate(self, instance, value): instance_value = None if value is None: self.error(instance, value) elif isinstance(value, string_types): try: prop = getattr( self.instance_class, '_props', {} ).get(self.uid_prop) if prop: value = prop.validate(None, value) if hasattr(self.instance_class, 'validate_uid'): self.instance_class.validate_uid(value) if self.load and hasattr(self.instance_class, 'load'): instance_value = self.instance_class.load(value) except utils.ValidationError as err: self.error(instance, value, extra=text_type(err)) else: instance_value = value if instance_value is not None: return super(Pointer, self).validate(instance, instance_value) return value def deserialize(self, value, **kwargs): """Deserialize instance from JSON value If a deserializer is registered, that is used. Otherwise, if the instance_class is a HasProperties subclass, an instance can be deserialized from a dictionary. """ kwargs.update({'trusted': kwargs.get('trusted', False)}) if self.deserializer is not None: return self.deserializer(value, **kwargs) if value is None: return None if isinstance(value, string_types): return value if issubclass(self.instance_class, base.HasProperties): return self.instance_class.deserialize(value, **kwargs) return self.from_json(value, **kwargs) def sphinx_class(self): """Description of the property, supplemental to the basic doc""" classdoc = super(Pointer, self).sphinx_class() return '{} instance or UID'.format(classdoc)