]> git.madduck.net Git - etc/taskwarrior.git/blobdiff - tasklib/task.py

madduck's git repository

Every one of the projects in this repository is available at the canonical URL git://git.madduck.net/madduck/pub/<projectpath> — see each project's metadata for the exact URL.

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.

SSH access, as well as push access can be individually arranged.

If you use my repositories frequently, consider adding the following snippet to ~/.gitconfig and using the third clone URL listed for each project:

[url "git://git.madduck.net/madduck/"]
  insteadOf = madduck:

projects / etc / taskwarrior.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
Taskwarrior 2.3.0 is the only one available for download now
[etc/taskwarrior.git] / tasklib / task.py
diff --git a/tasklib/task.py b/tasklib/task.py
index 5a223e222d91c70313d480be74441409a6e56a85..de0443caef2ad4732aafdbca31d393e860ac44ad 100644 (file)
--- a/tasklib/task.py
+++ b/tasklib/task.py
@@ -1,48 +1,80 @@
 import copy
 import datetime
 import json
 import copy
 import datetime
 import json
+import logging
 import os
 import subprocess
 import os
 import subprocess
-import tempfile
-import uuid
-
 
 DATE_FORMAT = '%Y%m%dT%H%M%SZ'
 
 DATE_FORMAT = '%Y%m%dT%H%M%SZ'
-
 REPR_OUTPUT_SIZE = 10
 REPR_OUTPUT_SIZE = 10
-
 PENDING = 'pending'
 PENDING = 'pending'
+COMPLETED = 'completed'
+
+logger = logging.getLogger(__name__)
 
 
 class TaskWarriorException(Exception):
     pass
 
 
 
 
 class TaskWarriorException(Exception):
     pass
 
 
-class Task(object):
+class TaskResource(object):
+    read_only_fields = []
 
 
-    class DoesNotExist(Exception):
-        pass
-
-    def __init__(self, warrior, data={}):
-        self.warrior = warrior
+    def _load_data(self, data):
         self._data = data
 
     def __getitem__(self, key):
         self._data = data
 
     def __getitem__(self, key):
-        return self._get_field(key)
+        hydrate_func = getattr(self, 'deserialize_{0}'.format(key),
+                               lambda x: x)
+        return hydrate_func(self._data.get(key))
+
+    def __setitem__(self, key, value):
+        if key in self.read_only_fields:
+            raise RuntimeError('Field \'%s\' is read-only' % key)
+        dehydrate_func = getattr(self, 'serialize_{0}'.format(key),
+                                 lambda x: x)
+        self._data[key] = dehydrate_func(value)
+        self._modified_fields.add(key)
+
+    def __repr__(self):
+        return self.__unicode__()
+
+
+class TaskAnnotation(TaskResource):
+    read_only_fields = ['entry', 'description']
+
+    def __init__(self, task, data={}):
+        self.task = task
+        self._load_data(data)
 
 
-    def __setitem__(self, key, val):
-        self._data[key] = val
+    def deserialize_entry(self, data):
+        return datetime.datetime.strptime(data, DATE_FORMAT) if data else None
+
+    def serialize_entry(self, date):
+        return date.strftime(DATE_FORMAT) if date else ''
+
+    def remove(self):
+        self.task.remove_annotation(self)
 
     def __unicode__(self):
 
     def __unicode__(self):
-        return self._data.get('description')
+        return self['description']
 
 
-    def _get_field(self, key):
-        hydrate_func = getattr(self, 'deserialize_{0}'.format(key), lambda x:x)
-        return hydrate_func(self._data.get(key))
+    __repr__ = __unicode__
 
 
-    def _set_field(self, key, value):
-        dehydrate_func = getattr(self, 'serialize_{0}'.format(key), lambda x:x)
-        self._data[key] = dehydrate_func(value)
+
+class Task(TaskResource):
+    read_only_fields = ['id', 'entry', 'urgency']
+
+    class DoesNotExist(Exception):
+        pass
+
+    def __init__(self, warrior, data={}):
+        self.warrior = warrior
+        self._load_data(data)
+        self._modified_fields = set()
+
+    def __unicode__(self):
+        return self['description']
 
     def serialize_due(self, date):
         return date.strftime(DATE_FORMAT)
 
     def serialize_due(self, date):
         return date.strftime(DATE_FORMAT)
@@ -52,36 +84,58 @@ class Task(object):
             return None
         return datetime.datetime.strptime(date_str, DATE_FORMAT)
 
             return None
         return datetime.datetime.strptime(date_str, DATE_FORMAT)
 
-    def serialize_annotations(self, annotations):
-        ann_list = list(annotations)
-        for ann in ann_list:
-            ann['entry'] = ann['entry'].strftime(DATE_FORMAT)
-        return ann_list
+    def deserialize_annotations(self, data):
+        return [TaskAnnotation(self, d) for d in data] if data else []
 
 
-    def deserialize_annotations(self, annotations):
-        ann_list = list(annotations)
-        for ann in ann_list:
-            ann['entry'] = datetime.datetime.strptime(
-                ann['entry'], DATE_FORMAT)
-        return ann_list
+    def deserialize_tags(self, tags):
+        if isinstance(tags, basestring):
+            return tags.split(',') if tags else []
+        return tags
 
 
-    def regenerate_uuid(self):
-        self['uuid'] = str(uuid.uuid4())
+    def serialize_tags(self, tags):
+        return ','.join(tags) if tags else ''
 
     def delete(self):
 
     def delete(self):
-        self.warrior.delete_task(self['uuid'])
+        self.warrior.execute_command([self['id'], 'delete'], config_override={
+            'confirmation': 'no',
+        })
 
     def done(self):
 
     def done(self):
-        self.warrior.complete_task(self['uuid'])
-
-    def save(self, delete_first=True):
-        if self['uuid'] and delete_first:
-            self.delete()
-        if not self['uuid'] or delete_first:
-            self.regenerate_uuid()
-        self.warrior.import_tasks([self._data])
-
-    __repr__ = __unicode__
+        self.warrior.execute_command([self['id'], 'done'])
+
+    def save(self):
+        args = [self['id'], 'modify'] if self['id'] else ['add']
+        args.extend(self._get_modified_fields_as_args())
+        self.warrior.execute_command(args)
+        self._modified_fields.clear()
+
+    def add_annotation(self, annotation):
+        args = [self['id'], 'annotate', annotation]
+        self.warrior.execute_command(args)
+        self.refresh(only_fields=['annotations'])
+
+    def remove_annotation(self, annotation):
+        if isinstance(annotation, TaskAnnotation):
+            annotation = annotation['description']
+        args = [self['id'], 'denotate', annotation]
+        self.warrior.execute_command(args)
+        self.refresh(only_fields=['annotations'])
+
+    def _get_modified_fields_as_args(self):
+        args = []
+        for field in self._modified_fields:
+            args.append('{}:{}'.format(field, self._data[field]))
+        return args
+
+    def refresh(self, only_fields=[]):
+        args = [self['uuid'], 'export']
+        new_data = json.loads(self.warrior.execute_command(args)[0])
+        if only_fields:
+            to_update = dict(
+                [(k, new_data.get(k)) for k in only_fields])
+            self._data.update(to_update)
+        else:
+            self._data = new_data
 
 
 class TaskFilter(object):
 
 
 class TaskFilter(object):
@@ -123,8 +177,8 @@ class TaskQuerySet(object):
         Deep copy of a QuerySet doesn't populate the cache
         """
         obj = self.__class__()
         Deep copy of a QuerySet doesn't populate the cache
         """
         obj = self.__class__()
-        for k,v in self.__dict__.items():
-            if k in ('_iter','_result_cache'):
+        for k, v in self.__dict__.items():
+            if k in ('_iter', '_result_cache'):
                 obj.__dict__[k] = None
             else:
                 obj.__dict__[k] = copy.deepcopy(v, memo)
                 obj.__dict__[k] = None
             else:
                 obj.__dict__[k] = copy.deepcopy(v, memo)
@@ -186,6 +240,9 @@ class TaskQuerySet(object):
     def pending(self):
         return self.filter(status=PENDING)
 
     def pending(self):
         return self.filter(status=PENDING)
 
+    def completed(self):
+        return self.filter(status=COMPLETED)
+
     def filter(self, *args, **kwargs):
         """
         Returns a new TaskQuerySet with the given filters added.
     def filter(self, *args, **kwargs):
         """
         Returns a new TaskQuerySet with the given filters added.
@@ -217,7 +274,8 @@ class TaskQuerySet(object):
 
 class TaskWarrior(object):
     def __init__(self, data_location='~/.task', create=True):
 
 class TaskWarrior(object):
     def __init__(self, data_location='~/.task', create=True):
-        if not os.path.exists(data_location):
+        data_location = os.path.expanduser(data_location)
+        if create and not os.path.exists(data_location):
             os.makedirs(data_location)
         self.config = {
             'data.location': os.path.expanduser(data_location),
             os.makedirs(data_location)
         self.config = {
             'data.location': os.path.expanduser(data_location),
@@ -230,17 +288,21 @@ class TaskWarrior(object):
         config.update(config_override)
         for item in config.items():
             command_args.append('rc.{0}={1}'.format(*item))
         config.update(config_override)
         for item in config.items():
             command_args.append('rc.{0}={1}'.format(*item))
-        command_args.extend(args)
+        command_args.extend(map(str, args))
         return command_args
 
     def execute_command(self, args, config_override={}):
         command_args = self._get_command_args(
             args, config_override=config_override)
         return command_args
 
     def execute_command(self, args, config_override={}):
         command_args = self._get_command_args(
             args, config_override=config_override)
+        logger.debug(' '.join(command_args))
         p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE)
         p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE)
-        stdout, stderr = p.communicate()
+        stdout, stderr = [x.decode() for x in p.communicate()]
         if p.returncode:
         if p.returncode:
-            error_msg = stderr.strip().splitlines()[-1]
+            if stderr.strip():
+                error_msg = stderr.strip().splitlines()[-1]
+            else:
+                error_msg = stdout.strip()
             raise TaskWarriorException(error_msg)
         return stdout.strip().split('\n')
 
             raise TaskWarriorException(error_msg)
         return stdout.strip().split('\n')
 
@@ -249,33 +311,20 @@ class TaskWarrior(object):
         tasks = []
         for line in self.execute_command(args):
             if line:
         tasks = []
         for line in self.execute_command(args):
             if line:
-                tasks.append(Task(self, json.loads(line.strip(','))))
+                data = line.strip(',')
+                try:
+                    tasks.append(Task(self, json.loads(data)))
+                except ValueError:
+                    raise TaskWarriorException('Invalid JSON: %s' % data)
         return tasks
 
         return tasks
 
-    def add_task(self, description, project=None):
-        args = ['add', description]
-        if project is not None:
-            args.append('project:{0}'.format(project))
-        self.execute_command(args)
-
-    def delete_task(self, task_id):
-        args = [task_id, 'rc.confirmation:no', 'delete']
-        self.execute_command(args)
-
-    def complete_task(self, task_id):
-        args = [task_id, 'done']
-        self.execute_command(args)
-
-    def import_tasks(self, tasks):
-        fd, path = tempfile.mkstemp()
-        with open(path, 'w') as f:
-            f.write(json.dumps(tasks))
-        args = ['import', path]
-        self.execute_command(args)
-
     def merge_with(self, path, push=False):
         path = path.rstrip('/') + '/'
     def merge_with(self, path, push=False):
         path = path.rstrip('/') + '/'
-        args = ['merge', path]
-        self.execute_command(args, config_override={
+        self.execute_command(['merge', path], config_override={
             'merge.autopush': 'yes' if push else 'no',
         })
             'merge.autopush': 'yes' if push else 'no',
         })
+
+    def undo(self):
+        self.execute_command(['undo'], config_override={
+            'confirmation': 'no',
+        })

Legalese

All content available in this repository is copyright © martin f. krafft, unless otherwise noted.

All files and their content are released under the terms of the Artistic Licence 2.0, unless otherwise noted.

The operator of this repository shall not be held liable for offending content in pages linking to or linked from these page, nor does he endorse their content unless stated otherwise.