]> git.madduck.net Git - etc/taskwarrior.git/blobdiff - tasklib/backends.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:

Don't run `_get_version` when testing custom command
[etc/taskwarrior.git] / tasklib / backends.py
index dfef0a873328653ac15051b5039065432254a4fb..38f6c597087dceaf9615a87b197a812a25f75f19 100644 (file)
@@ -1,4 +1,5 @@
 import abc
+import copy
 import datetime
 import json
 import logging
@@ -7,7 +8,7 @@ import re
 import six
 import subprocess
 
-from .task import Task, TaskQuerySet
+from .task import Task, TaskQuerySet, ReadOnlyDictView
 from .filters import TaskWarriorFilter
 from .serializing import local_zone
 
@@ -15,6 +16,7 @@ DATE_FORMAT_CALC = '%Y-%m-%dT%H:%M:%S'
 
 logger = logging.getLogger(__name__)
 
+
 class Backend(object):
 
     @abc.abstractproperty
@@ -73,14 +75,14 @@ class Backend(object):
         Converts TW syntax datetime string to a localized datetime
         object. This method is not mandatory.
         """
-        raise NotImplemented
+        raise NotImplementedError
 
 
 class TaskWarriorException(Exception):
     pass
 
 
-class TaskWarrior(object):
+class TaskWarrior(Backend):
 
     VERSION_2_1_0 = six.u('2.1.0')
     VERSION_2_2_0 = six.u('2.2.0')
@@ -92,16 +94,23 @@ class TaskWarrior(object):
     VERSION_2_4_4 = six.u('2.4.4')
     VERSION_2_4_5 = six.u('2.4.5')
 
-    def __init__(self, data_location=None, create=True, taskrc_location='~/.taskrc'):
-        self.taskrc_location = os.path.expanduser(taskrc_location)
+    def __init__(self, data_location=None, create=True,
+                 taskrc_location=None, task_command='task',
+                 version_override=None):
+        self.taskrc_location = None
+        if taskrc_location:
+            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 = '/'
 
-        # 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.task_command = task_command
 
-        self.version = self._get_version()
-        self.config = {
+        self._config = None
+        self.version = version_override or self._get_version()
+        self.overrides = {
             'confirmation': 'no',
             'dependency.confirmation': 'no',  # See TW-1483 or taskrc man page
             'recurrence.confirmation': 'no',  # Necessary for modifying R tasks
@@ -119,24 +128,30 @@ class TaskWarrior(object):
             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.overrides['data.location'] = data_location
 
         self.tasks = TaskQuerySet(self)
 
+    def _get_task_command(self):
+        return self.task_command.split()
+
     def _get_command_args(self, args, config_override=None):
-        command_args = ['task', 'rc:{0}'.format(self.taskrc_location)]
-        config = self.config.copy()
-        config.update(config_override or dict())
-        for item in config.items():
+        command_args = self._get_task_command()
+        overrides = self.overrides.copy()
+        overrides.update(config_override or dict())
+        for item in overrides.items():
             command_args.append('rc.{0}={1}'.format(*item))
-        command_args.extend(map(six.text_type, args))
+        command_args.extend([
+            x.decode('utf-8') if isinstance(x, six.binary_type)
+            else six.text_type(x) for x in args
+        ])
         return command_args
 
     def _get_version(self):
         p = subprocess.Popen(
-                ['task', '--version'],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE)
+            self._get_task_command() + ['--version'],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE)
         stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
         return stdout.strip('\n')
 
@@ -153,10 +168,11 @@ class TaskWarrior(object):
             if serialized_value is '':
                 escaped_serialized_value = ''
             else:
-                escaped_serialized_value = six.u("'{0}'").format(serialized_value)
+                escaped_serialized_value = six.u("'{0}'").format(
+                    serialized_value)
 
-            format_default = lambda task: six.u("{0}:{1}").format(field,
-                                                      escaped_serialized_value)
+            format_default = lambda task: six.u("{0}:{1}").format(
+                field, escaped_serialized_value)
 
             format_func = getattr(self, 'format_{0}'.format(field),
                                   format_default)
@@ -167,11 +183,17 @@ class TaskWarrior(object):
         if task.saved:
             for field in task._modified_fields:
                 add_field(field)
+
         # For new tasks, pass all fields that make sense
         else:
             for field in task._data.keys():
+                # We cannot set stuff that's read only (ID, UUID, ..)
                 if field in task.read_only_fields:
                     continue
+                # We do not want to do field deletion for new tasks
+                if task._data[field] is None:
+                    continue
+                # Otherwise we're fine
                 add_field(field)
 
         return args
@@ -191,9 +213,9 @@ class TaskWarrior(object):
 
         # Removed dependencies need to be prefixed with '-'
         return 'depends:' + ','.join(
-                [t['uuid'] for t in added] +
-                ['-' + t['uuid'] for t in removed]
-            )
+            [t['uuid'] for t in added] +
+            ['-' + t['uuid'] for t in removed]
+        )
 
     def format_description(self, task):
         # Task version older than 2.4.0 ignores first word of the
@@ -201,21 +223,27 @@ class TaskWarrior(object):
         if self.version < self.VERSION_2_4_0:
             return task._data['description']
         else:
-            return six.u("description:'{0}'").format(task._data['description'] or '')
+            return six.u("description:'{0}'").format(
+                task._data['description'] or '',
+            )
 
     def convert_datetime_string(self, value):
 
         if self.version >= self.VERSION_2_4_0:
-            # For strings, use 'task calc' to evaluate the string to datetime
+            # For strings, use 'calc' to evaluate the string to datetime
             # available since TW 2.4.0
             args = value.split()
             result = self.execute_command(['calc'] + args)
             naive = datetime.datetime.strptime(result[0], DATE_FORMAT_CALC)
             localized = local_zone.localize(naive)
         else:
-            raise ValueError("Provided value could not be converted to "
-                             "datetime, its type is not supported: {}"
-                             .format(type(value)))
+            raise ValueError(
+                'Provided value could not be converted to '
+                'datetime, its type is not supported: {}'
+                .format(type(value)),
+            )
+
+        return localized
 
     @property
     def filter_class(self):
@@ -223,35 +251,49 @@ class TaskWarrior(object):
 
     # Public interface
 
-    def get_config(self):
+    @property
+    def config(self):
+        # First, check if memoized information is available
+        if self._config:
+            return self._config
+
+        # If not, fetch the config using the 'show' command
         raw_output = self.execute_command(
-                ['show'],
-                config_override={'verbose': 'nothing'}
-            )
+            ['show'],
+            config_override={'verbose': 'nothing'}
+        )
 
         config = dict()
-        config_regex = re.compile(r'^(?P<key>[^\s]+)\s+(?P<value>[^\s].+$)')
+        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
+        # Memoize the config dict
+        self._config = ReadOnlyDictView(config)
+
+        return self._config
 
     def execute_command(self, args, config_override=None, allow_failure=True,
                         return_all=False):
         command_args = self._get_command_args(
             args, config_override=config_override)
-        logger.debug(' '.join(command_args))
+        logger.debug(u' '.join(command_args))
+
+        env = os.environ.copy()
+        if self.taskrc_location:
+            env['TASKRC'] = self.taskrc_location
         p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
-                             stderr=subprocess.PIPE)
+                             stderr=subprocess.PIPE, env=env)
         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()
+            error_msg += u'\nCommand used: ' + u' '.join(command_args)
             raise TaskWarriorException(error_msg)
 
         # Return all whole triplet only if explicitly asked for
@@ -283,7 +325,7 @@ class TaskWarrior(object):
 
     def filter_tasks(self, filter_obj):
         self.enforce_recurrence()
-        args = ['export', '--'] + filter_obj.get_filter_params()
+        args = ['export'] + filter_obj.get_filter_params()
         tasks = []
         for line in self.execute_command(args):
             if line:
@@ -308,10 +350,14 @@ class TaskWarrior(object):
             id_lines = [l for l in output if l.startswith('Created task ')]
 
             # Complain loudly if it seems that more tasks were created
-            # Should not happen
-            if len(id_lines) != 1 or len(id_lines[0].split(' ')) != 3:
-                raise TaskWarriorException("Unexpected output when creating "
-                                           "task: %s" % '\n'.join(id_lines))
+            # Should not happen.
+            # Expected output: Created task 1.
+            #                  Created task 1 (recurrence template).
+            if len(id_lines) != 1 or len(id_lines[0].split(' ')) not in (3, 5):
+                raise TaskWarriorException(
+                    'Unexpected output when creating '
+                    'task: %s' % '\n'.join(id_lines),
+                )
 
             # Circumvent the ID storage, since ID is considered read-only
             identifier = id_lines[0].split(' ')[2].rstrip('.')
@@ -377,15 +423,15 @@ class TaskWarrior(object):
             for key, value in data.items():
                 taskfilter.add_filter_param(key, value)
 
-            output = self.execute_command(['export', '--'] +
-                taskfilter.get_filter_params())
+            output = self.execute_command(['export'] +
+                                          taskfilter.get_filter_params())
 
         # If more than 1 task has been matched still, raise an exception
         if not valid(output):
             raise TaskWarriorException(
-                "Unique identifiers {0} with description: {1} matches "
-                "multiple tasks: {2}".format(
-                task['uuid'] or task['id'], task['description'], output)
+                'Unique identifiers {0} with description: {1} matches '
+                'multiple tasks: {2}'.format(
+                    task['uuid'] or task['id'], task['description'], output)
             )
 
         return json.loads(output[0])