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

backends: Move TaskWarriorException into backends
[etc/taskwarrior.git] / tasklib / task.py
index 35fc2b710c7d99fde6cd2397a02fbe442aa2a0ec..4f10b9dc358f21c2cb9d671f3bae5706aceb57a2 100644 (file)
@@ -5,34 +5,22 @@ import json
 import logging
 import os
 import pytz
-import re
 import six
 import sys
-import subprocess
 import tzlocal
 
+from backends import TaskWarrior, TaskWarriorException
+
 DATE_FORMAT = '%Y%m%dT%H%M%SZ'
 DATE_FORMAT_CALC = '%Y-%m-%dT%H:%M:%S'
 REPR_OUTPUT_SIZE = 10
 PENDING = 'pending'
 COMPLETED = 'completed'
 
-VERSION_2_1_0 = six.u('2.1.0')
-VERSION_2_2_0 = six.u('2.2.0')
-VERSION_2_3_0 = six.u('2.3.0')
-VERSION_2_4_0 = six.u('2.4.0')
-VERSION_2_4_1 = six.u('2.4.1')
-VERSION_2_4_2 = six.u('2.4.2')
-VERSION_2_4_3 = six.u('2.4.3')
-
 logger = logging.getLogger(__name__)
 local_zone = tzlocal.get_localzone()
 
 
-class TaskWarriorException(Exception):
-    pass
-
-
 class ReadOnlyDictView(object):
     """
     Provides simplified read-only view upon dict object.
@@ -392,9 +380,9 @@ class TaskResource(SerializingObject):
 class TaskAnnotation(TaskResource):
     read_only_fields = ['entry', 'description']
 
-    def __init__(self, task, data={}):
+    def __init__(self, task, data=None):
         self.task = task
-        self._load_data(data)
+        self._load_data(data or dict())
         super(TaskAnnotation, self).__init__(task.warrior)
 
     def remove(self):
@@ -761,7 +749,7 @@ class Task(TaskResource):
         # For older TW versions attempt to uniquely locate the task
         # using the data we have if it has been just saved.
         # This can happen when adding a completed task on older TW versions.
-        if (not valid(output) and self.warrior.version < six.text_type(u'2.4.5')
+        if (not valid(output) and self.warrior.version < VERSION_2_4_5
                 and after_save):
 
             # Make a copy, removing ID and UUID. It's most likely invalid
@@ -826,8 +814,10 @@ class TaskFilter(SerializingObject):
 
             # We enforce equality match by using 'is' (or 'none') modifier
             # Without using this syntax, filter fails due to TW-1479
-            modifier = '.is' if value else '.none'
-            key = key + modifier if '.' not in key else key
+            # which is, however, fixed in 2.4.5
+            if self.warrior.version < VERSION_2_4_5:
+                modifier = '.is' if value else '.none'
+                key = key + modifier if '.' not in key else key
 
             self.filter_params.append(six.u("{0}:{1}").format(key, value))
 
@@ -948,124 +938,3 @@ class TaskQuerySet(object):
         raise ValueError(
             'get() returned more than one Task -- it returned {0}! '
             'Lookup parameters were {1}'.format(num, kwargs))
-
-
-class TaskWarrior(object):
-    def __init__(self, data_location=None, create=True, taskrc_location='~/.taskrc'):
-        self.taskrc_location = os.path.expanduser(taskrc_location)
-
-        # If taskrc does not exist, pass / to use defaults and avoid creating
-        # dummy .taskrc file by TaskWarrior
-        if not os.path.exists(self.taskrc_location):
-            self.taskrc_location = '/'
-
-        self.version = self._get_version()
-        self.config = {
-            'confirmation': 'no',
-            'dependency.confirmation': 'no',  # See TW-1483 or taskrc man page
-            'recurrence.confirmation': 'no',  # Necessary for modifying R tasks
-
-            # Defaults to on since 2.4.5, we expect off during parsing
-            'json.array': 'off',
-
-            # 2.4.3 onwards supports 0 as infite bulk, otherwise set just
-            # arbitrary big number which is likely to be large enough
-            'bulk': 0 if self.version >= VERSION_2_4_3 else 100000,
-        }
-
-        # Set data.location override if passed via kwarg
-        if data_location is not None:
-            data_location = os.path.expanduser(data_location)
-            if create and not os.path.exists(data_location):
-                os.makedirs(data_location)
-            self.config['data.location'] = data_location
-
-        self.tasks = TaskQuerySet(self)
-
-    def _get_command_args(self, args, config_override={}):
-        command_args = ['task', 'rc:{0}'.format(self.taskrc_location)]
-        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(six.text_type, args))
-        return command_args
-
-    def _get_version(self):
-        p = subprocess.Popen(
-                ['task', '--version'],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE)
-        stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
-        return stdout.strip('\n')
-
-    def get_config(self):
-        raw_output = self.execute_command(
-                ['show'],
-                config_override={'verbose': 'nothing'}
-            )
-
-        config = dict()
-        config_regex = re.compile(r'^(?P<key>[^\s]+)\s+(?P<value>[^\s].+$)')
-
-        for line in raw_output:
-            match = config_regex.match(line)
-            if match:
-                config[match.group('key')] = match.group('value').strip()
-
-        return config
-
-    def execute_command(self, args, config_override={}, allow_failure=True,
-                        return_all=False):
-        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 and allow_failure:
-            if stderr.strip():
-                error_msg = stderr.strip()
-            else:
-                error_msg = stdout.strip()
-            raise TaskWarriorException(error_msg)
-
-        # Return all whole triplet only if explicitly asked for
-        if not return_all:
-            return stdout.rstrip().split('\n')
-        else:
-            return (stdout.rstrip().split('\n'),
-                    stderr.rstrip().split('\n'),
-                    p.returncode)
-
-    def enforce_recurrence(self):
-        # Run arbitrary report command which will trigger generation
-        # of recurrent tasks.
-
-        # Only necessary for TW up to 2.4.1, fixed in 2.4.2.
-        if self.version < VERSION_2_4_2:
-            self.execute_command(['next'], allow_failure=False)
-
-    def filter_tasks(self, filter_obj):
-        self.enforce_recurrence()
-        args = ['export', '--'] + filter_obj.get_filter_params()
-        tasks = []
-        for line in self.execute_command(args):
-            if line:
-                data = line.strip(',')
-                try:
-                    filtered_task = Task(self)
-                    filtered_task._load_data(json.loads(data))
-                    tasks.append(filtered_task)
-                except ValueError:
-                    raise TaskWarriorException('Invalid JSON: %s' % data)
-        return tasks
-
-    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 undo(self):
-        self.execute_command(['undo'])