]> 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
Bump version for 0.4.0
[etc/taskwarrior.git] / tasklib / task.py
diff --git a/tasklib/task.py b/tasklib/task.py
index cd4b92f1d3af491bc001229406ac26f3e509c5a7..6fbce776d658bef73b250983d86042273cb28e57 100644 (file)
--- a/tasklib/task.py
+++ b/tasklib/task.py
@@ -1,54 +1,146 @@
+from __future__ import print_function
 import copy
 import copy
+import datetime
 import json
 import json
+import logging
 import os
 import os
+import six
 import subprocess
 import subprocess
-import tempfile
-import uuid
-
 
 
+DATE_FORMAT = '%Y%m%dT%H%M%SZ'
 REPR_OUTPUT_SIZE = 10
 PENDING = 'pending'
 REPR_OUTPUT_SIZE = 10
 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 = []
+
+    def _load_data(self, data):
+        self._data = data
+
+    def __getitem__(self, 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 __str__(self):
+        s = six.text_type(self.__unicode__())
+        if not six.PY3:
+            s = s.encode('utf-8')
+        return s
+
+
+class TaskAnnotation(TaskResource):
+    read_only_fields = ['entry', 'description']
+
+    def __init__(self, task, data={}):
+        self.task = task
+        self._load_data(data)
+
+    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):
+        return self['description']
+
+    __repr__ = __unicode__
+
+
+class Task(TaskResource):
+    read_only_fields = ['id', 'entry', 'urgency']
 
     class DoesNotExist(Exception):
         pass
 
     def __init__(self, warrior, data={}):
         self.warrior = warrior
 
     class DoesNotExist(Exception):
         pass
 
     def __init__(self, warrior, data={}):
         self.warrior = warrior
-        self._data = data
+        self._load_data(data)
+        self._modified_fields = set()
 
 
-    def __getitem__(self, key):
-        return self._data.get(key)
+    def __unicode__(self):
+        return self['description']
 
 
-    def __setitem__(self, key, val):
-        self._data[key] = val
+    def serialize_due(self, date):
+        return date.strftime(DATE_FORMAT)
 
 
-    def __unicode__(self):
-        return self._data.get('description')
+    def deserialize_due(self, date_str):
+        if not date_str:
+            return None
+        return datetime.datetime.strptime(date_str, DATE_FORMAT)
 
 
-    def regenerate_uuid(self):
-        self['uuid'] = str(uuid.uuid4())
+    def deserialize_annotations(self, data):
+        return [TaskAnnotation(self, d) for d in data] if data else []
 
 
-    def delete(self):
-        self.warrior.delete_task(self['uuid'])
+    def deserialize_tags(self, tags):
+        if isinstance(tags, basestring):
+            return tags.split(',') if tags else []
+        return tags
 
 
-    def done(self):
-        self.warrior.complete_task(self['uuid'])
+    def serialize_tags(self, tags):
+        return ','.join(tags) if tags else ''
 
 
-    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])
+    def delete(self):
+        self.warrior.execute_command([self['id'], 'delete'], config_override={
+            'confirmation': 'no',
+        })
 
 
-    __repr__ = __unicode__
+    def done(self):
+        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):
@@ -56,18 +148,22 @@ class TaskFilter(object):
     A set of parameters to filter the task list with.
     """
 
     A set of parameters to filter the task list with.
     """
 
-    def __init__(self, filter_params=()):
+    def __init__(self, filter_params=[]):
         self.filter_params = filter_params
 
         self.filter_params = filter_params
 
-    def add_filter(self, param, value):
-        self.filter_params += ((param, value),)
+    def add_filter(self, filter_str):
+        self.filter_params.append(filter_str)
+
+    def add_filter_param(self, key, value):
+        key = key.replace('__', '.')
+        self.filter_params.append('{0}:{1}'.format(key, value))
 
     def get_filter_params(self):
 
     def get_filter_params(self):
-        return self.filter_params
+        return [f for f in self.filter_params if f]
 
     def clone(self):
         c = self.__class__()
 
     def clone(self):
         c = self.__class__()
-        c.filter_params = tuple(self.filter_params)
+        c.filter_params = list(self.filter_params)
         return c
 
 
         return c
 
 
@@ -86,8 +182,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)
@@ -138,7 +234,7 @@ class TaskQuerySet(object):
         """
         Fetch the tasks which match the current filters.
         """
         """
         Fetch the tasks which match the current filters.
         """
-        return self.warrior._execute_filter(self.filter_obj)
+        return self.warrior.filter_tasks(self.filter_obj)
 
     def all(self):
         """
 
     def all(self):
         """
@@ -149,13 +245,18 @@ class TaskQuerySet(object):
     def pending(self):
         return self.filter(status=PENDING)
 
     def pending(self):
         return self.filter(status=PENDING)
 
-    def filter(self, **kwargs):
+    def completed(self):
+        return self.filter(status=COMPLETED)
+
+    def filter(self, *args, **kwargs):
         """
         Returns a new TaskQuerySet with the given filters added.
         """
         clone = self._clone()
         """
         Returns a new TaskQuerySet with the given filters added.
         """
         clone = self._clone()
-        for param, value in kwargs.items():
-            clone.filter_obj.add_filter(param, value)
+        for f in args:
+            clone.filter_obj.add_filter(f)
+        for key, value in kwargs.items():
+            clone.filter_obj.add_filter_param(key, value)
         return clone
 
     def get(self, **kwargs):
         return clone
 
     def get(self, **kwargs):
@@ -177,62 +278,58 @@ class TaskQuerySet(object):
 
 
 class TaskWarrior(object):
 
 
 class TaskWarrior(object):
-    DEFAULT_FILTERS = {
-        'status': 'pending',
-    }
-
     def __init__(self, data_location='~/.task', create=True):
     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),
         }
         self.tasks = TaskQuerySet(self)
 
             os.makedirs(data_location)
         self.config = {
             'data.location': os.path.expanduser(data_location),
         }
         self.tasks = TaskQuerySet(self)
 
-    def _generate_command(self, command):
-        args = ['task', 'rc:/']
-        for item in self.config.items():
-            args.append('rc.{0}={1}'.format(*item))
-        args.append(command)
-        return ' '.join(args)
-
-    def _execute_command(self, command):
-        p = subprocess.Popen(self._generate_command(command), shell=True,
-                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        stdout, stderr = p.communicate()
+    def _get_command_args(self, args, config_override={}):
+        command_args = ['task', 'rc:/']
+        config = self.config.copy()
+        config.update(config_override)
+        for item in config.items():
+            command_args.append('rc.{0}={1}'.format(*item))
+        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)
+        logger.debug(' '.join(command_args))
+        p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
         if p.returncode:
         if p.returncode:
-            raise TaskWarriorException(stderr.strip())
+            if stderr.strip():
+                error_msg = stderr.strip().splitlines()[-1]
+            else:
+                error_msg = stdout.strip()
+            raise TaskWarriorException(error_msg)
         return stdout.strip().split('\n')
 
         return stdout.strip().split('\n')
 
-    def _format_filter_kwarg(self, kwarg):
-        key, val = kwarg[0], kwarg[1]
-        key = key.replace('__', '.')
-        return '{0}:{1}'.format(key, val)
-
-    def _execute_filter(self, filter_obj):
-        filter_commands = ' '.join(map(self._format_filter_kwarg,
-                                       filter_obj.get_filter_params()))
-        command = '{0} export'.format(filter_commands)
+    def filter_tasks(self, filter_obj):
+        args = ['export', '--'] + filter_obj.get_filter_params()
         tasks = []
         tasks = []
-        for line in self._execute_command(command):
+        for line in self.execute_command(args):
             if line:
             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(' '.join(args))
-
-    def delete_task(self, task_id):
-        self._execute_command('{0} rc.confirmation:no delete'.format(task_id))
-
-    def complete_task(self, task_id):
-        self._execute_command('{0} done'.format(task_id))
+    def merge_with(self, path, push=False):
+        path = path.rstrip('/') + '/'
+        self.execute_command(['merge', path], config_override={
+            'merge.autopush': 'yes' if push else 'no',
+        })
 
 
-    def import_tasks(self, tasks):
-        fd, path = tempfile.mkstemp()
-        with open(path, 'w') as f:
-            f.write(json.dumps(tasks))
-        self._execute_command('import {0}'.format(path))
+    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.