Source code for ftrack_api.attribute

# :coding: utf-8
# :copyright: Copyright (c) 2014 ftrack

from __future__ import absolute_import

from builtins import object
import collections
from six.moves import collections_abc
import copy
import logging
import functools

import ftrack_api.symbol
import ftrack_api.exception
import ftrack_api.collection
import ftrack_api.inspection
import ftrack_api.operation

logger = logging.getLogger(
    __name__
)


[docs]def merge_references(function): '''Decorator to handle merging of references / collections.''' @functools.wraps(function) def get_value(attribute, entity): '''Merge the attribute with the local cache.''' if attribute.name not in entity._inflated: # Only merge on first access to avoid # inflating them multiple times. logger.debug( 'Merging potential new data into attached ' 'entity for attribute {0}.'.format( attribute.name ) ) # Local attributes. local_value = attribute.get_local_value(entity) if isinstance( local_value, ( ftrack_api.entity.base.Entity, ftrack_api.collection.Collection, ftrack_api.collection.MappedCollectionProxy ) ): logger.debug( 'Merging local value for attribute {0}.'.format(attribute) ) merged_local_value = entity.session._merge( local_value, merged=dict() ) if merged_local_value is not local_value: with entity.session.operation_recording(False): attribute.set_local_value(entity, merged_local_value) # Remote attributes. remote_value = attribute.get_remote_value(entity) if isinstance( remote_value, ( ftrack_api.entity.base.Entity, ftrack_api.collection.Collection, ftrack_api.collection.MappedCollectionProxy ) ): logger.debug( 'Merging remote value for attribute {0}.'.format(attribute) ) merged_remote_value = entity.session._merge( remote_value, merged=dict() ) if merged_remote_value is not remote_value: attribute.set_remote_value(entity, merged_remote_value) entity._inflated.add( attribute.name ) return function( attribute, entity ) return get_value
[docs]class Attributes(object): '''Collection of properties accessible by name.'''
[docs] def __init__(self, attributes=None): super(Attributes, self).__init__() self._data = dict() if attributes is not None: for attribute in attributes: self.add(attribute)
[docs] def add(self, attribute): '''Add *attribute*.''' existing = self._data.get(attribute.name, None) if existing: raise ftrack_api.exception.NotUniqueError( 'Attribute with name {0} already added as {1}' .format(attribute.name, existing) ) self._data[attribute.name] = attribute
[docs] def remove(self, attribute): '''Remove attribute.''' self._data.pop(attribute.name)
[docs] def get(self, name): '''Return attribute by *name*. If no attribute matches *name* then return None. ''' return self._data.get(name, None)
[docs] def keys(self): '''Return list of attribute names.''' return list(self._data.keys())
def __contains__(self, item): '''Return whether *item* present.''' if not isinstance(item, Attribute): return False return item.name in self._data def __iter__(self): '''Return iterator over attributes.''' return iter(self._data.values()) def __len__(self): '''Return count of attributes.''' return len(self._data)
[docs]class Attribute(object): '''A name and value pair persisted remotely.'''
[docs] def __init__( self, name, default_value=ftrack_api.symbol.NOT_SET, mutable=True, computed=False ): '''Initialise attribute with *name*. *default_value* represents the default value for the attribute. It may be a callable. It is not used within the attribute when providing values, but instead exists for other parts of the system to reference. If *mutable* is set to False then the local value of the attribute on an entity can only be set when both the existing local and remote values are :attr:`ftrack_api.symbol.NOT_SET`. The exception to this is when the target value is also :attr:`ftrack_api.symbol.NOT_SET`. If *computed* is set to True the value is a remote side computed value and should not be long-term cached. ''' super(Attribute, self).__init__() self._name = name self._mutable = mutable self._computed = computed self.default_value = default_value self._local_key = 'local' self._remote_key = 'remote'
def __repr__(self): '''Return representation of entity.''' return '<{0}.{1}({2}) object at {3}>'.format( self.__module__, self.__class__.__name__, self.name, id(self) )
[docs] def get_entity_storage(self, entity): '''Return attribute storage on *entity* creating if missing.''' storage_key = '_ftrack_attribute_storage' storage = getattr(entity, storage_key, None) if storage is None: storage = collections.defaultdict( lambda: { self._local_key: ftrack_api.symbol.NOT_SET, self._remote_key: ftrack_api.symbol.NOT_SET } ) setattr(entity, storage_key, storage) return storage
@property def name(self): '''Return name.''' return self._name @property def mutable(self): '''Return whether attribute is mutable.''' return self._mutable @property def computed(self): '''Return whether attribute is computed.''' return self._computed
[docs] def get_value(self, entity): '''Return current value for *entity*. If a value was set locally then return it, otherwise return last known remote value. If no remote value yet retrieved, make a request for it via the session and block until available. ''' value = self.get_local_value(entity) if value is not ftrack_api.symbol.NOT_SET: return value value = self.get_remote_value(entity) if value is not ftrack_api.symbol.NOT_SET: return value if not entity.session.auto_populate: return value self.populate_remote_value(entity) return self.get_remote_value(entity)
[docs] def get_local_value(self, entity): '''Return locally set value for *entity*.''' storage = self.get_entity_storage(entity) return storage[self.name][self._local_key]
[docs] def get_remote_value(self, entity): '''Return remote value for *entity*. .. note:: Only return locally stored remote value, do not fetch from remote. ''' storage = self.get_entity_storage(entity) return storage[self.name][self._remote_key]
[docs] def set_local_value(self, entity, value): '''Set local *value* for *entity*.''' if ( not self.mutable and self.is_set(entity) and value is not ftrack_api.symbol.NOT_SET ): raise ftrack_api.exception.ImmutableAttributeError(self) old_value = self.get_local_value(entity) storage = self.get_entity_storage(entity) storage[self.name][self._local_key] = value # Record operation. if entity.session.record_operations: entity.session.recorded_operations.push( ftrack_api.operation.UpdateEntityOperation( entity.entity_type, ftrack_api.inspection.primary_key(entity), self.name, old_value, value ) )
[docs] def set_remote_value(self, entity, value): '''Set remote *value*. .. note:: Only set locally stored remote value, do not persist to remote. ''' storage = self.get_entity_storage(entity) storage[self.name][self._remote_key] = value
[docs] def populate_remote_value(self, entity): '''Populate remote value for *entity*.''' entity.session.populate([entity], self.name)
[docs] def is_modified(self, entity): '''Return whether local value set and differs from remote. .. note:: Will not fetch remote value so may report True even when values are the same on the remote. ''' local_value = self.get_local_value(entity) remote_value = self.get_remote_value(entity) return ( local_value is not ftrack_api.symbol.NOT_SET and local_value != remote_value )
[docs] def is_set(self, entity): '''Return whether a value is set for *entity*.''' return any([ self.get_local_value(entity) is not ftrack_api.symbol.NOT_SET, self.get_remote_value(entity) is not ftrack_api.symbol.NOT_SET ])
[docs]class ScalarAttribute(Attribute): '''Represent a scalar value.'''
[docs] def __init__(self, name, data_type, **kw): '''Initialise property.''' super(ScalarAttribute, self).__init__(name, **kw) self.data_type = data_type
[docs]class ReferenceAttribute(Attribute): '''Reference another entity.'''
[docs] def __init__(self, name, entity_type, **kw): '''Initialise property.''' super(ReferenceAttribute, self).__init__(name, **kw) self.entity_type = entity_type
[docs] def populate_remote_value(self, entity): '''Populate remote value for *entity*. As attribute references another entity, use that entity's configured default projections to auto populate useful attributes when loading. ''' reference_entity_type = entity.session.types[self.entity_type] default_projections = reference_entity_type.default_projections projections = [] if default_projections: for projection in default_projections: projections.append('{0}.{1}'.format(self.name, projection)) else: projections.append(self.name) entity.session.populate([entity], ', '.join(projections))
[docs] def is_modified(self, entity): '''Return whether a local value has been set and differs from remote. .. note:: Will not fetch remote value so may report True even when values are the same on the remote. ''' local_value = self.get_local_value(entity) remote_value = self.get_remote_value(entity) if local_value is ftrack_api.symbol.NOT_SET: return False if remote_value is ftrack_api.symbol.NOT_SET: return True if ( ftrack_api.inspection.identity(local_value) != ftrack_api.inspection.identity(remote_value) ): return True return False
[docs] @merge_references def get_value(self, entity): return super(ReferenceAttribute, self).get_value( entity )
[docs]class AbstractCollectionAttribute(Attribute): '''Base class for collection attributes.''' #: Collection class used by attribute. collection_class = None
[docs] @merge_references def get_value(self, entity): '''Return current value for *entity*. If a value was set locally then return it, otherwise return last known remote value. If no remote value yet retrieved, make a request for it via the session and block until available. .. note:: As value is a collection that is mutable, will transfer a remote value into the local value on access if no local value currently set. ''' super(AbstractCollectionAttribute, self).get_value(entity) # Conditionally, copy remote value into local value so that it can be # mutated without side effects. local_value = self.get_local_value(entity) remote_value = self.get_remote_value(entity) if ( local_value is ftrack_api.symbol.NOT_SET and isinstance(remote_value, self.collection_class) ): try: with entity.session.operation_recording(False): self.set_local_value(entity, copy.copy(remote_value)) except ftrack_api.exception.ImmutableAttributeError: pass value = self.get_local_value(entity) # If the local value is still not set then attempt to set it with a # suitable placeholder collection so that the caller can interact with # the collection using its normal interface. This is required for a # newly created entity for example. It *could* be done as a simple # default value, but that would incur cost for every collection even # when they are not modified before commit. if value is ftrack_api.symbol.NOT_SET: try: with entity.session.operation_recording(False): self.set_local_value( entity, # None should be treated as empty collection. None ) except ftrack_api.exception.ImmutableAttributeError: pass return self.get_local_value(entity)
[docs] def set_local_value(self, entity, value): '''Set local *value* for *entity*.''' if value is not ftrack_api.symbol.NOT_SET: value = self._adapt_to_collection(entity, value) value.mutable = self.mutable super(AbstractCollectionAttribute, self).set_local_value(entity, value)
[docs] def set_remote_value(self, entity, value): '''Set remote *value*. .. note:: Only set locally stored remote value, do not persist to remote. ''' if value is not ftrack_api.symbol.NOT_SET: value = self._adapt_to_collection(entity, value) value.mutable = False super(AbstractCollectionAttribute, self).set_remote_value(entity, value)
def _adapt_to_collection(self, entity, value): '''Adapt *value* to appropriate collection instance for *entity*. .. note:: If *value* is None then return a suitable empty collection. ''' raise NotImplementedError()
[docs]class CollectionAttribute(AbstractCollectionAttribute): '''Represent a collection of other entities.''' #: Collection class used by attribute. collection_class = ftrack_api.collection.Collection def _adapt_to_collection(self, entity, value): '''Adapt *value* to a Collection instance on *entity*.''' if not isinstance(value, ftrack_api.collection.Collection): if value is None: value = ftrack_api.collection.Collection(entity, self) elif isinstance(value, list): value = ftrack_api.collection.Collection( entity, self, data=value ) else: raise NotImplementedError( 'Cannot convert {0!r} to collection.'.format(value) ) else: if value.attribute is not self: raise ftrack_api.exception.AttributeError( 'Collection already bound to a different attribute' ) return value
[docs]class KeyValueMappedCollectionAttribute(AbstractCollectionAttribute): '''Represent a mapped key, value collection of entities.''' #: Collection class used by attribute. collection_class = ftrack_api.collection.KeyValueMappedCollectionProxy
[docs] def __init__( self, name, creator, key_attribute, value_attribute, **kw ): '''Initialise attribute with *name*. *creator* should be a function that accepts a dictionary of data and is used by the referenced collection to create new entities in the collection. *key_attribute* should be the name of the attribute on an entity in the collection that represents the value for 'key' of the dictionary. *value_attribute* should be the name of the attribute on an entity in the collection that represents the value for 'value' of the dictionary. ''' self.creator = creator self.key_attribute = key_attribute self.value_attribute = value_attribute super(KeyValueMappedCollectionAttribute, self).__init__(name, **kw)
def _adapt_to_collection(self, entity, value): '''Adapt *value* to an *entity*.''' if not isinstance( value, ftrack_api.collection.KeyValueMappedCollectionProxy ): if value is None: value = ftrack_api.collection.KeyValueMappedCollectionProxy( ftrack_api.collection.Collection(entity, self), self.creator, self.key_attribute, self.value_attribute ) elif isinstance(value, (list, ftrack_api.collection.Collection)): if isinstance(value, list): value = ftrack_api.collection.Collection( entity, self, data=value ) value = ftrack_api.collection.KeyValueMappedCollectionProxy( value, self.creator, self.key_attribute, self.value_attribute ) elif isinstance(value, collections_abc.Mapping): # Convert mapping. # TODO: When backend model improves, revisit this logic. # First get existing value and delete all references. This is # needed because otherwise they will not be automatically # removed server side. # The following should not cause recursion as the internal # values should be mapped collections already. current_value = self.get_value(entity) if not isinstance( current_value, ftrack_api.collection.KeyValueMappedCollectionProxy ): raise NotImplementedError( 'Cannot adapt mapping to collection as current value ' 'type is not a KeyValueMappedCollectionProxy.' ) # Create the new collection using the existing collection as # basis. Then update through proxy interface to ensure all # internal operations called consistently (such as entity # deletion for key removal). collection = ftrack_api.collection.Collection( entity, self, data=current_value.collection[:] ) collection_proxy = ( ftrack_api.collection.KeyValueMappedCollectionProxy( collection, self.creator, self.key_attribute, self.value_attribute ) ) # Remove expired keys from collection. expired_keys = set(current_value.keys()) - set(value.keys()) for key in expired_keys: del collection_proxy[key] # Set new values for existing keys / add new keys. for key, value in list(value.items()): collection_proxy[key] = value value = collection_proxy else: raise NotImplementedError( 'Cannot convert {0!r} to collection.'.format(value) ) else: if value.attribute is not self: raise ftrack_api.exception.AttributeError( 'Collection already bound to a different attribute.' ) return value
[docs]class CustomAttributeCollectionAttribute(AbstractCollectionAttribute): '''Represent a mapped custom attribute collection of entities.''' #: Collection class used by attribute. collection_class = ( ftrack_api.collection.CustomAttributeCollectionProxy ) def _adapt_to_collection(self, entity, value): '''Adapt *value* to an *entity*.''' if not isinstance( value, ftrack_api.collection.CustomAttributeCollectionProxy ): if value is None: value = ftrack_api.collection.CustomAttributeCollectionProxy( ftrack_api.collection.Collection(entity, self) ) elif isinstance(value, (list, ftrack_api.collection.Collection)): # Why are we creating a new if it is a list? This will cause # any merge to create a new proxy and collection. if isinstance(value, list): value = ftrack_api.collection.Collection( entity, self, data=value ) value = ftrack_api.collection.CustomAttributeCollectionProxy( value ) elif isinstance(value, collections_abc.Mapping): # Convert mapping. # TODO: When backend model improves, revisit this logic. # First get existing value and delete all references. This is # needed because otherwise they will not be automatically # removed server side. # The following should not cause recursion as the internal # values should be mapped collections already. current_value = self.get_value(entity) if not isinstance( current_value, ftrack_api.collection.CustomAttributeCollectionProxy ): raise NotImplementedError( 'Cannot adapt mapping to collection as current value ' 'type is not a MappedCollectionProxy.' ) # Create the new collection using the existing collection as # basis. Then update through proxy interface to ensure all # internal operations called consistently (such as entity # deletion for key removal). collection = ftrack_api.collection.Collection( entity, self, data=current_value.collection[:] ) collection_proxy = ( ftrack_api.collection.CustomAttributeCollectionProxy( collection ) ) # Remove expired keys from collection. expired_keys = set(current_value.keys()) - set(value.keys()) for key in expired_keys: del collection_proxy[key] # Set new values for existing keys / add new keys. for key, value in list(value.items()): collection_proxy[key] = value value = collection_proxy else: raise NotImplementedError( 'Cannot convert {0!r} to collection.'.format(value) ) else: if value.attribute is not self: raise ftrack_api.exception.AttributeError( 'Collection already bound to a different attribute.' ) return value