]>
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:
import re
import six
import subprocess
import re
import six
import subprocess
from .task import Task, TaskQuerySet, ReadOnlyDictView
from .filters import TaskWarriorFilter
from .task import Task, TaskQuerySet, ReadOnlyDictView
from .filters import TaskWarriorFilter
Converts TW syntax datetime string to a localized datetime
object. This method is not mandatory.
"""
Converts TW syntax datetime string to a localized datetime
object. This method is not mandatory.
"""
+ raise NotImplementedError
class TaskWarriorException(Exception):
class TaskWarriorException(Exception):
overrides.update(config_override or dict())
for item in overrides.items():
command_args.append('rc.{0}={1}'.format(*item))
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):
return command_args
def _get_version(self):
if task.saved:
for field in task._modified_fields:
add_field(field)
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():
# 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
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
add_field(field)
return args
- 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)
for line in raw_output:
match = config_regex.match(line)
return_all=False):
command_args = self._get_command_args(
args, config_override=config_override)
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))
+
p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
p = subprocess.Popen(command_args, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
error_msg = stderr.strip()
else:
error_msg = stdout.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
raise TaskWarriorException(error_msg)
# Return all whole triplet only if explicitly asked for
def filter_tasks(self, filter_obj):
self.enforce_recurrence()
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:
tasks = []
for line in self.execute_command(args):
if line:
id_lines = [l for l in output if l.startswith('Created task ')]
# Complain loudly if it seems that more tasks were created
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:
+ # 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))
raise TaskWarriorException("Unexpected output when creating "
"task: %s" % '\n'.join(id_lines))
for key, value in data.items():
taskfilter.add_filter_param(key, value)
for key, value in data.items():
taskfilter.add_filter_param(key, value)
- output = self.execute_command(['export', '--' ] +
+ output = self.execute_command(['export'] +
taskfilter.get_filter_params())
# If more than 1 task has been matched still, raise an exception
taskfilter.get_filter_params())
# If more than 1 task has been matched still, raise an exception
projects / etc / taskwarrior.git / blobdiff