import logging
import os
import six
+import sys
import subprocess
DATE_FORMAT = '%Y%m%dT%H%M%SZ'
return ','.join(tags) if tags else ''
def deserialize_tags(self, tags):
- if isinstance(tags, basestring):
+ if isinstance(tags, six.string_types):
return tags.split(',') if tags else []
return tags or []
# are not propagated.
self._original_data = copy.deepcopy(self._data)
+ def _update_data(self, data, update_original=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()))
+
+ 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
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__
"""
pass
+ @classmethod
+ def from_input(cls, input_file=sys.stdin, modify=None):
+ """
+ Creates a Task object, directly from the stdin, by reading one line.
+ If modify=True, two lines are used, first line interpreted as the
+ original state of the Task object, and second line as its new,
+ modified value. This is consistent with the TaskWarrior's hook
+ system.
+
+ Object created by this method should not be saved, deleted
+ or refreshed, as t could create a infinite loop. For this
+ reason, TaskWarrior instance is set to None.
+
+ Input_file argument can be used to specify the input file,
+ but defaults to sys.stdin.
+ """
+
+ # TaskWarrior instance is set to None
+ task = cls(None)
+
+ # Detect the hook type if not given directly
+ name = os.path.basename(sys.argv[0])
+ modify = name.startswith('on-modify') if modify is None else modify
+
+ # Load the data from the input
+ task._load_data(json.loads(input_file.readline().strip()))
+
+ # If this is a on-modify event, we are provided with additional
+ # line of input, which provides updated data
+ if modify:
+ task._update_data(json.loads(input_file.readline().strip()))
+
+ return task
+
def __init__(self, warrior, **kwargs):
self.warrior = warrior
def _modified_fields(self):
writable_fields = set(self._data.keys()) - set(self.read_only_fields)
for key in writable_fields:
- if self._data.get(key) != self._original_data.get(key):
+ 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 _is_modified(self):
+ def modified(self):
return bool(list(self._modified_fields))
@property
raise Task.NotSaved("Task needs to be saved before it can be deleted")
# Refresh the status, and raise exception if the task is deleted
- self.refresh()
+ self.refresh(only_fields=['status'])
if self.deleted:
raise Task.DeletedTask("Task was already deleted")
self.warrior.execute_command([self['uuid'], 'delete'])
# Refresh the status again, so that we have updated info stored
- self.refresh()
+ self.refresh(only_fields=['status'])
def done(self):
raise Task.NotSaved("Task needs to be saved before it can be completed")
# Refresh, and raise exception if task is already completed/deleted
- self.refresh()
+ self.refresh(only_fields=['status'])
if self.completed:
raise Task.CompletedTask("Cannot complete a completed task")
self.warrior.execute_command([self['uuid'], 'done'])
# Refresh the status again, so that we have updated info stored
- self.refresh()
+ self.refresh(only_fields=['status'])
def save(self):
- if self.saved and not self._is_modified:
+ if self.saved and not self.modified:
return
args = [self['uuid'], 'modify'] if self.saved else ['add']
# Circumvent the ID storage, since ID is considered read-only
self._data['id'] = int(id_lines[0].split(' ')[2].rstrip('.'))
+ # Refreshing is very important here, as not only modification time
+ # is updated, but arbitrary attribute may have changed due hooks
+ # altering the data before saving
self.refresh()
def add_annotation(self, annotation):
args = [self['uuid'], 'annotate', annotation]
self.warrior.execute_command(args)
- self.refresh()
+ self.refresh(only_fields=['annotations'])
def remove_annotation(self, annotation):
if not self.saved:
annotation = annotation['description']
args = [self['uuid'], 'denotate', annotation]
self.warrior.execute_command(args)
- self.refresh()
+ self.refresh(only_fields=['annotations'])
def _get_modified_fields_as_args(self):
args = []
# Add the output of format_field method to args list (defaults to
# field:value)
serialized_value = self._serialize(field, self._data[field]) or ''
- format_default = lambda: "{0}:'{1}'".format(field, serialized_value)
+ format_default = lambda: "{0}:{1}".format(
+ field,
+ "'{0}'".format(serialized_value) if serialized_value else ''
+ )
format_func = getattr(self, 'format_{0}'.format(field),
format_default)
args.append(format_func())
return args
- def refresh(self):
+ def refresh(self, only_fields=[]):
# Raise error when trying to refresh a task that has not been saved
if not self.saved:
raise Task.NotSaved("Task needs to be saved to be refreshed")
# with using UUID only.
args = [self['uuid'] or self['id'], 'export']
new_data = json.loads(self.warrior.execute_command(args)[0])
- self._load_data(new_data)
+ if only_fields:
+ to_update = dict(
+ [(k, new_data.get(k)) for k in only_fields])
+ self._update_data(to_update, update_original=True)
+ else:
+ self._load_data(new_data)
+
+ 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=(',',':'))
class TaskFilter(SerializingObject):
"""
projects / etc / taskwarrior.git / blobdiff