pass
+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 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 SerializingObject(object):
"""
Common ancestor for TaskResource & TaskFilter, since they both
not export empty-valued attributes) if the attribute
is not iterable (e.g. list or set), in which case
a empty iterable should be used.
+
+ Normalizing methods should hold the following contract:
+ - They are used to validate and normalize the user input.
+ Any attribute value that comes from the user (during Task
+ initialization, assignign values to Task attributes, or
+ filtering by user-provided values of attributes) is first
+ validated and normalized using the normalize_{key} method.
+ - If validation or normalization fails, normalizer is expected
+ to raise ValueError.
"""
def _deserialize(self, key, value):
or entered as a value of Task attribute.
"""
+ # None value should not be converted by normalizer
+ if value is None:
+ return None
+
normalize_func = getattr(self, 'normalize_{0}'.format(key),
lambda x: x)
def normalize_modified(self, value):
return self.datetime_normalizer(value)
+ def serialize_start(self, value):
+ return self.timestamp_serializer(value)
+
+ def deserialize_start(self, value):
+ return self.timestamp_deserializer(value)
+
+ def normalize_start(self, value):
+ return self.datetime_normalizer(value)
+
+ def serialize_end(self, value):
+ return self.timestamp_serializer(value)
+
+ def deserialize_end(self, value):
+ return self.timestamp_deserializer(value)
+
+ def normalize_end(self, value):
+ return self.datetime_normalizer(value)
+
def serialize_due(self, value):
return self.timestamp_serializer(value)
# If the value is already localized, there is no need to change
# time zone at this point. Also None is a valid value too.
localized = value
-
+
return localized
def normalize_uuid(self, value):
# Enforce sane UUID
- if not isinstance(value, six.text_type) or value == '':
- raise ValueError("UUID must be a valid non-empty string.")
+ if not isinstance(value, six.string_types) or value == '':
+ raise ValueError("UUID must be a valid non-empty string, "
+ "not: {}".format(value))
return value
pass
@classmethod
- def from_input(cls, input_file=sys.stdin, modify=None):
+ def from_input(cls, input_file=sys.stdin, modify=None, warrior=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
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
+ # Create the TaskWarrior instance if none passed
+ if warrior is None:
+ hook_parent_dir = os.path.dirname(os.path.dirname(sys.argv[0]))
+ warrior = TaskWarrior(data_location=hook_parent_dir)
+
+ # TaskWarrior instance is set to None
+ task = cls(warrior)
+
# Load the data from the input
task._load_data(json.loads(input_file.readline().strip()))
for (key, value) in six.iteritems(kwargs))
self._original_data = copy.deepcopy(self._data)
+ # Provide read only access to the original data
+ self.original = ReadOnlyDictView(self._original_data)
+
def __unicode__(self):
return self['description']
self.warrior.execute_command([self['uuid'], 'delete'])
# Refresh the status again, so that we have updated info stored
+ self.refresh(only_fields=['status', 'start', 'end'])
+
+ def start(self):
+ if not self.saved:
+ raise Task.NotSaved("Task needs to be saved before it can be started")
+
+ # Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
+ if self.completed:
+ raise Task.CompletedTask("Cannot start a completed task")
+ elif self.deleted:
+ raise Task.DeletedTask("Deleted task cannot be started")
+
+ self.warrior.execute_command([self['uuid'], 'start'])
+
+ # Refresh the status again, so that we have updated info stored
+ self.refresh(only_fields=['status', 'start'])
def done(self):
if not self.saved:
self.warrior.execute_command([self['uuid'], 'done'])
# Refresh the status again, so that we have updated info stored
- self.refresh(only_fields=['status'])
+ self.refresh(only_fields=['status', 'start', 'end'])
def save(self):
if self.saved and not self.modified:
stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
if p.returncode and allow_failure:
if stderr.strip():
- error_msg = stderr.strip().splitlines()[-1]
+ error_msg = stderr.strip()
else:
error_msg = stdout.strip()
raise TaskWarriorException(error_msg)
projects / etc / taskwarrior.git / blobdiff