]> 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:

Fix mismatched parentheses bug
[etc/taskwarrior.git] / tasklib / task.py
index 5e2b71b66a48d4b87e6ad8c27ccbb932bb76ace4..4681418448b1ca18b77a5bf5b69f2f9850ebf9f4 100644 (file)
@@ -1,3 +1,5 @@
+import copy
+import datetime
 import json
 import os
 import subprocess
@@ -5,6 +7,10 @@ import tempfile
 import uuid
 
 
+DATE_FORMAT = '%Y%m%dT%H%M%SZ'
+
+REPR_OUTPUT_SIZE = 10
+
 PENDING = 'pending'
 
 
@@ -14,12 +20,15 @@ class TaskWarriorException(Exception):
 
 class Task(object):
 
+    class DoesNotExist(Exception):
+        pass
+
     def __init__(self, warrior, data={}):
         self.warrior = warrior
         self._data = data
 
     def __getitem__(self, key):
-        return self._data.get(key)
+        return self._get_field(key)
 
     def __setitem__(self, key, val):
         self._data[key] = val
@@ -27,6 +36,35 @@ class Task(object):
     def __unicode__(self):
         return self._data.get('description')
 
+    def _get_field(self, key):
+        hydrate_func = getattr(self, 'deserialize_{0}'.format(key), lambda x:x)
+        return hydrate_func(self._data.get(key))
+
+    def _set_field(self, key, value):
+        dehydrate_func = getattr(self, 'serialize_{0}'.format(key), lambda x:x)
+        self._data[key] = dehydrate_func(value)
+
+    def serialize_due(self, date):
+        return date.strftime(DATE_FORMAT)
+
+    def deserialize_due(self, date_str):
+        if not date_str:
+            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, annotations):
+        ann_list = list(annotations)
+        for ann in ann_list:
+            ann['entry'] = datetime.datetime.strptime(
+                ann['entry'], DATE_FORMAT)
+        return ann_list
+
     def regenerate_uuid(self):
         self['uuid'] = str(uuid.uuid4())
 
@@ -46,6 +84,137 @@ class Task(object):
     __repr__ = __unicode__
 
 
+class TaskFilter(object):
+    """
+    A set of parameters to filter the task list with.
+    """
+
+    def __init__(self, filter_params=[]):
+        self.filter_params = filter_params
+
+    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):
+        return [f for f in self.filter_params if f]
+
+    def clone(self):
+        c = self.__class__()
+        c.filter_params = list(self.filter_params)
+        return c
+
+
+class TaskQuerySet(object):
+    """
+    Represents a lazy lookup for a task objects.
+    """
+
+    def __init__(self, warrior=None, filter_obj=None):
+        self.warrior = warrior
+        self._result_cache = None
+        self.filter_obj = filter_obj or TaskFilter()
+
+    def __deepcopy__(self, memo):
+        """
+        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'):
+                obj.__dict__[k] = None
+            else:
+                obj.__dict__[k] = copy.deepcopy(v, memo)
+        return obj
+
+    def __repr__(self):
+        data = list(self[:REPR_OUTPUT_SIZE + 1])
+        if len(data) > REPR_OUTPUT_SIZE:
+            data[-1] = "...(remaining elements truncated)..."
+        return repr(data)
+
+    def __len__(self):
+        if self._result_cache is None:
+            self._result_cache = list(self)
+        return len(self._result_cache)
+
+    def __iter__(self):
+        if self._result_cache is None:
+            self._result_cache = self._execute()
+        return iter(self._result_cache)
+
+    def __getitem__(self, k):
+        if self._result_cache is None:
+            self._result_cache = list(self)
+        return self._result_cache.__getitem__(k)
+
+    def __bool__(self):
+        if self._result_cache is not None:
+            return bool(self._result_cache)
+        try:
+            next(iter(self))
+        except StopIteration:
+            return False
+        return True
+
+    def __nonzero__(self):
+        return type(self).__bool__(self)
+
+    def _clone(self, klass=None, **kwargs):
+        if klass is None:
+            klass = self.__class__
+        filter_obj = self.filter_obj.clone()
+        c = klass(warrior=self.warrior, filter_obj=filter_obj)
+        c.__dict__.update(kwargs)
+        return c
+
+    def _execute(self):
+        """
+        Fetch the tasks which match the current filters.
+        """
+        return self.warrior.filter_tasks(self.filter_obj)
+
+    def all(self):
+        """
+        Returns a new TaskQuerySet that is a copy of the current one.
+        """
+        return self._clone()
+
+    def pending(self):
+        return self.filter(status=PENDING)
+
+    def filter(self, *args, **kwargs):
+        """
+        Returns a new TaskQuerySet with the given filters added.
+        """
+        clone = self._clone()
+        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):
+        """
+        Performs the query and returns a single object matching the given
+        keyword arguments.
+        """
+        clone = self.filter(**kwargs)
+        num = len(clone)
+        if num == 1:
+            return clone._result_cache[0]
+        if not num:
+            raise Task.DoesNotExist(
+                'Task matching query does not exist. '
+                'Lookup parameters were {0}'.format(kwargs))
+        raise ValueError(
+            'get() returned more than one Task -- it returned {0}! '
+            'Lookup parameters were {1}'.format(num, kwargs))
+
+
 class TaskWarrior(object):
     DEFAULT_FILTERS = {
         'status': 'pending',
@@ -57,59 +226,48 @@ class TaskWarrior(object):
         self.config = {
             'data.location': os.path.expanduser(data_location),
         }
+        self.tasks = TaskQuerySet(self)
 
-    def _generate_command(self, command):
-        args = ['task', 'rc:/']
+    def _get_command_args(self, args):
+        command_args = ['task', 'rc:/']
         for item in self.config.items():
-            args.append('rc.{0}={1}'.format(*item))
-        args.append(command)
-        return ' '.join(args)
+            command_args.append('rc.{0}={1}'.format(*item))
+        command_args.extend(args)
+        return command_args
 
-    def _execute(self, command):
-        p = subprocess.Popen(self._generate_command(command), shell=True,
+    def _execute_command(self, args):
+        p = subprocess.Popen(self._get_command_args(args),
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         stdout, stderr = p.communicate()
         if p.returncode:
             raise TaskWarriorException(stderr.strip())
         return stdout.strip().split('\n')
 
-    def _format_filter_kwarg(self, kwarg):
-        key, val = kwarg[0], kwarg[1]
-        if key in ['tag', 'tags']:
-            key = 'tags.equal'
-        key = key.replace('__', '.')
-        return '{0}:{1}'.format(key, val)
-
-    def get_tasks(self, **filter_kwargs):
-        filters = self.DEFAULT_FILTERS
-        filters.update(filter_kwargs)
-        filter_commands = ' '.join(map(self._format_filter_kwarg,
-                                       filters.items()))
-        command = '{0} export'.format(filter_commands)
+    def filter_tasks(self, filter_obj):
+        args = ['export', '--'] + filter_obj.get_filter_params()
         tasks = []
-        for line in self._execute(command):
+        for line in self._execute_command(args):
             if line:
                 tasks.append(Task(self, json.loads(line.strip(','))))
         return tasks
 
-    def get_task(self, task_id):
-        command = '{0} export'.format(task_id)
-        return Task(self, json.loads(self._execute(command)[0]))
-
     def add_task(self, description, project=None):
         args = ['add', description]
         if project is not None:
             args.append('project:{0}'.format(project))
-        self._execute(' '.join(args))
+        self._execute_command(args)
 
     def delete_task(self, task_id):
-        self._execute('{0} rc.confirmation:no delete'.format(task_id))
+        args = [task_id, 'rc.confirmation:no', 'delete']
+        self._execute_command(args)
 
     def complete_task(self, task_id):
-        self._execute('{0} done'.format(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))
-        self._execute('import {0}'.format(path))
+        args = ['import', path]
+        self._execute_command(args)