+class ReadOnlyDictView(object):
+ """
+ Provides simplified read-only view upon dict object.
+ """
+
+ def __init__(self, viewed_dict):
+ self.viewed_dict = viewed_dict
+
+ def __getitem__(self, key):
+ return copy.deepcopy(self.viewed_dict.__getitem__(key))
+
+ def __contains__(self, k):
+ return self.viewed_dict.__contains__(k)
+
+ def __iter__(self):
+ for value in self.viewed_dict:
+ yield copy.deepcopy(value)
+
+ def __len__(self):
+ return len(self.viewed_dict)
+
+ def __unicode__(self):
+ return six.u('ReadOnlyDictView: {0}'.format(repr(self.viewed_dict)))
+
+ __repr__ = __unicode__
+
+ def get(self, key, default=None):
+ return copy.deepcopy(self.viewed_dict.get(key, default))
+
+ def items(self):
+ return [copy.deepcopy(v) for v in self.viewed_dict.items()]
+
+ def values(self):
+ return [copy.deepcopy(v) for v in self.viewed_dict.values()]
+
+
+class TaskResource(SerializingObject):
+ read_only_fields = []
+
+ def _load_data(self, data):
+ self._data = dict((key, self._deserialize(key, value))
+ for key, value in data.items())
+ # We need to use a copy for original data, so that changes
+ # are not propagated.
+ self._original_data = copy.deepcopy(self._data)
+
+ def _update_data(self, data, update_original=False, remove_missing=False):
+ """
+ Low level update of the internal _data dict. Data which are coming as
+ updates should already be serialized. If update_original is True, the
+ original_data dict is updated as well.
+ """
+ self._data.update(dict((key, self._deserialize(key, value))
+ for key, value in data.items()))
+
+ # In certain situations, we want to treat missing keys as removals
+ if remove_missing:
+ for key in set(self._data.keys()) - set(data.keys()):
+ self._data[key] = None
+
+ if update_original:
+ self._original_data = copy.deepcopy(self._data)
+
+ def __getitem__(self, key):
+ # This is a workaround to make TaskResource non-iterable
+ # over simple index-based iteration
+ try:
+ int(key)
+ raise StopIteration
+ except ValueError:
+ pass
+
+ if key not in self._data:
+ self._data[key] = self._deserialize(key, None)
+
+ return self._data.get(key)
+
+ def __setitem__(self, key, value):
+ if key in self.read_only_fields:
+ raise RuntimeError('Field \'%s\' is read-only' % key)
+
+ # Normalize the user input before saving it
+ value = self._normalize(key, value)
+ self._data[key] = value
+
+ def __str__(self):
+ s = six.text_type(self.__unicode__())
+ if not six.PY3:
+ s = s.encode('utf-8')
+ return s
+
+ def __repr__(self):
+ return str(self)
+
+ def export_data(self):
+ """
+ Exports current data contained in the Task as JSON
+ """
+
+ # We need to remove spaces for TW-1504, use custom separators
+ data_tuples = ((key, self._serialize(key, value))
+ for key, value in six.iteritems(self._data))
+
+ # Empty string denotes empty serialized value, we do not want
+ # to pass that to TaskWarrior.
+ data_tuples = filter(lambda t: t[1] is not '', data_tuples)
+ data = dict(data_tuples)
+ return json.dumps(data, separators=(',', ':'))
+
+ @property
+ def _modified_fields(self):
+ writable_fields = set(self._data.keys()) - set(self.read_only_fields)
+ for key in writable_fields:
+ new_value = self._data.get(key)
+ old_value = self._original_data.get(key)
+
+ # Make sure not to mark data removal as modified field if the
+ # field originally had some empty value
+ if key in self._data and not new_value and not old_value:
+ continue
+
+ if new_value != old_value:
+ yield key
+
+ @property
+ def modified(self):
+ return bool(list(self._modified_fields))
+
+
+class TaskAnnotation(TaskResource):
+ read_only_fields = ['entry', 'description']
+
+ def __init__(self, task, data=None):
+ self.task = task
+ self._load_data(data or dict())
+ super(TaskAnnotation, self).__init__(task.backend)
+
+ def remove(self):
+ self.task.remove_annotation(self)
+
+ def __unicode__(self):
+ return self['description']
+
+ def __eq__(self, other):
+ # consider 2 annotations equal if they belong to the same task, and
+ # their data dics are the same
+ return self.task == other.task and self._data == other._data
+
+ __repr__ = __unicode__
+
+
+class LazyUUIDTask(object):
+ """
+ A lazy wrapper around Task object, referenced by UUID.
+
+ - Supports comparison with LazyUUIDTask or Task objects (equality by UUIDs)
+ - If any attribute other than 'uuid' requested, a lookup in the
+ backend will be performed and this object will be replaced by a proper
+ Task object.
+ """
+
+ def __init__(self, tw, uuid):
+ self._tw = tw
+ self._uuid = uuid
+
+ def __getitem__(self, key):
+ # LazyUUIDTask does not provide anything else other than 'uuid'
+ if key is 'uuid':
+ return self._uuid
+ else:
+ self.replace()
+ return self[key]
+
+ def __getattr__(self, name):
+ # Getattr is called only if the attribute could not be found using
+ # normal means
+ self.replace()
+ return self.name
+
+ def __eq__(self, other):
+ if other['uuid']:
+ # For saved Tasks, just define equality by equality of uuids
+ return self['uuid'] == other['uuid']
+
+ def __hash__(self):
+ return self['uuid'].__hash__()
+
+ def replace(self):
+ """
+ Performs conversion to the regular Task object, referenced by the
+ stored UUID.
+ """
+
+ replacement = self._tw.tasks.get(uuid=self._uuid)
+ self.__class__ = replacement.__class__
+ self.__dict__ = replacement.__dict__
+
+
+class LazyUUIDTaskSet(object):
+ """
+ A lazy wrapper around TaskQuerySet object, for tasks referenced by UUID.
+
+ - Supports 'in' operator with LazyUUIDTask or Task objects
+ - If iteration over the objects in the LazyUUIDTaskSet is requested, the
+ LazyUUIDTaskSet will be converted to QuerySet and evaluated
+ """
+
+ def __init__(self, tw, uuids):
+ self._tw = tw
+ self._uuids = set(uuids)
+
+ def __getattr__(self, name):
+ # Getattr is called only if the attribute could not be found using
+ # normal means
+ self.replace()
+ return self.name
+
+ def __eq__(self, other):
+ return set(t['uuid'] for t in other) == self._uuids
+
+ def __contains__(self, task):
+ return task['uuid'] in self._uuids
+
+ def __len__(self):
+ return len(self._uuids)
+
+ def __iter__(self):
+ self.replace()
+ for task in self:
+ yield task
+
+ def replace(self):
+ """
+ Performs conversion to the regular TaskQuerySet object, referenced by
+ the stored UUIDs.
+ """
+
+ replacement = self._tw.tasks.filter(' '.join(self._uuids))
+ self.__class__ = replacement.__class__
+ self.__dict__ = replacement.__dict__
projects / etc / taskwarrior.git / blobdiff