All patches and comments are welcome. Please squash your changes to logical
commits before using git-format-patch and git-send-email to
patches@git.madduck.net.
If you'd read over the Git project's submission guidelines and adhered to them,
I'd be especially grateful.
1 from __future__ import print_function
10 DATE_FORMAT = '%Y%m%dT%H%M%SZ'
13 COMPLETED = 'completed'
15 logger = logging.getLogger(__name__)
18 class TaskWarriorException(Exception):
22 class TaskResource(object):
25 def _load_data(self, data):
28 def __getitem__(self, key):
29 hydrate_func = getattr(self, 'deserialize_{0}'.format(key),
31 return hydrate_func(self._data.get(key))
33 def __setitem__(self, key, value):
34 if key in self.read_only_fields:
35 raise RuntimeError('Field \'%s\' is read-only' % key)
36 dehydrate_func = getattr(self, 'serialize_{0}'.format(key),
38 self._data[key] = dehydrate_func(value)
39 self._modified_fields.add(key)
42 s = six.text_type(self.__unicode__())
51 class TaskAnnotation(TaskResource):
52 read_only_fields = ['entry', 'description']
54 def __init__(self, task, data={}):
58 def deserialize_entry(self, data):
59 return datetime.datetime.strptime(data, DATE_FORMAT) if data else None
61 def serialize_entry(self, date):
62 return date.strftime(DATE_FORMAT) if date else ''
65 self.task.remove_annotation(self)
67 def __unicode__(self):
68 return self['description']
70 __repr__ = __unicode__
73 class Task(TaskResource):
74 read_only_fields = ['id', 'entry', 'urgency', 'uuid']
76 class DoesNotExist(Exception):
79 class CompletedTask(Exception):
81 Raised when the operation cannot be performed on the completed task.
85 class DeletedTask(Exception):
87 Raised when the operation cannot be performed on the deleted task.
91 def __init__(self, warrior, data={}):
92 self.warrior = warrior
94 self._modified_fields = set()
96 def __unicode__(self):
97 return self['description']
101 return self['status'] == six.text_type('completed')
105 return self['status'] == six.text_type('deleted')
109 return self['status'] == six.text_type('waiting')
113 return self['status'] == six.text_type('pending')
117 return self['uuid'] is not None or self['id'] is not None
119 def serialize_due(self, date):
120 return date.strftime(DATE_FORMAT)
122 def deserialize_due(self, date_str):
125 return datetime.datetime.strptime(date_str, DATE_FORMAT)
127 def deserialize_annotations(self, data):
128 return [TaskAnnotation(self, d) for d in data] if data else []
130 def deserialize_tags(self, tags):
131 if isinstance(tags, basestring):
132 return tags.split(',') if tags else []
135 def serialize_tags(self, tags):
136 return ','.join(tags) if tags else ''
139 # Refresh the status, and raise exception if the task is deleted
140 self.refresh(only_fields=['status'])
143 raise self.DeletedTask("Task was already deleted")
145 self.warrior.execute_command([self['uuid'], 'delete'], config_override={
146 'confirmation': 'no',
149 # Refresh the status again, so that we have updated info stored
150 self.refresh(only_fields=['status'])
154 # Refresh, and raise exception if task is already completed/deleted
155 self.refresh(only_fields=['status'])
158 raise self.CompletedTask("Cannot complete a completed task")
160 raise self.DeletedTask("Deleted task cannot be completed")
162 self.warrior.execute_command([self['uuid'], 'done'])
164 # Refresh the status again, so that we have updated info stored
165 self.refresh(only_fields=['status'])
168 args = [self['uuid'], 'modify'] if self.saved else ['add']
169 args.extend(self._get_modified_fields_as_args())
170 output = self.warrior.execute_command(args)
172 # Parse out the new ID, if the task is being added for the first time
174 id_lines = [l for l in output if l.startswith('Created task ')]
176 # Complain loudly if it seems that more tasks were created
178 if len(id_lines) != 1 or len(id_lines[0].split(' ')) != 3:
179 raise TaskWarriorException("Unexpected output when creating "
180 "task: %s" % '\n'.join(id_lines))
182 # Circumvent the ID storage, since ID is considered read-only
183 self._data['id'] = int(id_lines[0].split(' ')[2].rstrip('.'))
185 self._modified_fields.clear()
188 def add_annotation(self, annotation):
189 args = [self['uuid'], 'annotate', annotation]
190 self.warrior.execute_command(args)
191 # TODO: This will not work with the tasks that are not yet saved
192 self.refresh(only_fields=['annotations'])
194 def remove_annotation(self, annotation):
195 if isinstance(annotation, TaskAnnotation):
196 annotation = annotation['description']
197 args = [self['uuid'], 'denotate', annotation]
198 self.warrior.execute_command(args)
199 # TODO: This will not work with the tasks that are not yet saved
200 self.refresh(only_fields=['annotations'])
202 def _get_modified_fields_as_args(self):
204 for field in self._modified_fields:
205 args.append('{}:{}'.format(field, self._data[field]))
208 def refresh(self, only_fields=[]):
209 # Do not refresh for tasks that are not yet saved in the TW
213 # We need to use ID as backup for uuid here for the refreshes
214 # of newly saved tasks. Any other place in the code is fine
215 # with using UUID only.
216 args = [self['uuid'] or self['id'], 'export']
217 new_data = json.loads(self.warrior.execute_command(args)[0])
220 [(k, new_data.get(k)) for k in only_fields])
221 self._data.update(to_update)
223 self._data = new_data
226 class TaskFilter(object):
228 A set of parameters to filter the task list with.
231 def __init__(self, filter_params=[]):
232 self.filter_params = filter_params
234 def add_filter(self, filter_str):
235 self.filter_params.append(filter_str)
237 def add_filter_param(self, key, value):
238 key = key.replace('__', '.')
240 # Replace the value with empty string, since that is the
241 # convention in TW for empty values
242 value = value if value is not None else ''
243 self.filter_params.append('{0}:{1}'.format(key, value))
245 def get_filter_params(self):
246 return [f for f in self.filter_params if f]
250 c.filter_params = list(self.filter_params)
254 class TaskQuerySet(object):
256 Represents a lazy lookup for a task objects.
259 def __init__(self, warrior=None, filter_obj=None):
260 self.warrior = warrior
261 self._result_cache = None
262 self.filter_obj = filter_obj or TaskFilter()
264 def __deepcopy__(self, memo):
266 Deep copy of a QuerySet doesn't populate the cache
268 obj = self.__class__()
269 for k, v in self.__dict__.items():
270 if k in ('_iter', '_result_cache'):
271 obj.__dict__[k] = None
273 obj.__dict__[k] = copy.deepcopy(v, memo)
277 data = list(self[:REPR_OUTPUT_SIZE + 1])
278 if len(data) > REPR_OUTPUT_SIZE:
279 data[-1] = "...(remaining elements truncated)..."
283 if self._result_cache is None:
284 self._result_cache = list(self)
285 return len(self._result_cache)
288 if self._result_cache is None:
289 self._result_cache = self._execute()
290 return iter(self._result_cache)
292 def __getitem__(self, k):
293 if self._result_cache is None:
294 self._result_cache = list(self)
295 return self._result_cache.__getitem__(k)
298 if self._result_cache is not None:
299 return bool(self._result_cache)
302 except StopIteration:
306 def __nonzero__(self):
307 return type(self).__bool__(self)
309 def _clone(self, klass=None, **kwargs):
311 klass = self.__class__
312 filter_obj = self.filter_obj.clone()
313 c = klass(warrior=self.warrior, filter_obj=filter_obj)
314 c.__dict__.update(kwargs)
319 Fetch the tasks which match the current filters.
321 return self.warrior.filter_tasks(self.filter_obj)
325 Returns a new TaskQuerySet that is a copy of the current one.
330 return self.filter(status=PENDING)
333 return self.filter(status=COMPLETED)
335 def filter(self, *args, **kwargs):
337 Returns a new TaskQuerySet with the given filters added.
339 clone = self._clone()
341 clone.filter_obj.add_filter(f)
342 for key, value in kwargs.items():
343 clone.filter_obj.add_filter_param(key, value)
346 def get(self, **kwargs):
348 Performs the query and returns a single object matching the given
351 clone = self.filter(**kwargs)
354 return clone._result_cache[0]
356 raise Task.DoesNotExist(
357 'Task matching query does not exist. '
358 'Lookup parameters were {0}'.format(kwargs))
360 'get() returned more than one Task -- it returned {0}! '
361 'Lookup parameters were {1}'.format(num, kwargs))
364 class TaskWarrior(object):
365 def __init__(self, data_location='~/.task', create=True):
366 data_location = os.path.expanduser(data_location)
367 if create and not os.path.exists(data_location):
368 os.makedirs(data_location)
370 'data.location': os.path.expanduser(data_location),
372 self.tasks = TaskQuerySet(self)
374 def _get_command_args(self, args, config_override={}):
375 command_args = ['task', 'rc:/']
376 config = self.config.copy()
377 config.update(config_override)
378 for item in config.items():
379 command_args.append('rc.{0}={1}'.format(*item))
380 command_args.extend(map(str, args))
383 def execute_command(self, args, config_override={}):
384 command_args = self._get_command_args(
385 args, config_override=config_override)
386 logger.debug(' '.join(command_args))
387 p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
388 stderr=subprocess.PIPE)
389 stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
392 error_msg = stderr.strip().splitlines()[-1]
394 error_msg = stdout.strip()
395 raise TaskWarriorException(error_msg)
396 return stdout.strip().split('\n')
398 def filter_tasks(self, filter_obj):
399 args = ['export', '--'] + filter_obj.get_filter_params()
401 for line in self.execute_command(args):
403 data = line.strip(',')
405 tasks.append(Task(self, json.loads(data)))
407 raise TaskWarriorException('Invalid JSON: %s' % data)
410 def merge_with(self, path, push=False):
411 path = path.rstrip('/') + '/'
412 self.execute_command(['merge', path], config_override={
413 'merge.autopush': 'yes' if push else 'no',
417 self.execute_command(['undo'], config_override={
418 'confirmation': 'no',