# :coding: utf-8
# :copyright: Copyright (c) 2014 ftrack
from __future__ import absolute_import
from builtins import object
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