]>
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):
VERSION_2_4_4 = six.u('2.4.4')
VERSION_2_4_5 = six.u('2.4.5')
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'):
+ 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
self.taskrc_location = os.path.expanduser(taskrc_location)
# If taskrc does not exist, pass / to use defaults and avoid creating
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
if self.version < self.VERSION_2_4_0:
return task._data['description']
else:
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):
def convert_datetime_string(self, value):
naive = datetime.datetime.strptime(result[0], DATE_FORMAT_CALC)
localized = local_zone.localize(naive)
else:
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)),
+ )
- 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:
- 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('.')
# Circumvent the ID storage, since ID is considered read-only
identifier = id_lines[0].split(' ')[2].rstrip('.')
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
if not valid(output):
raise TaskWarriorException(
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(
+ 'Unique identifiers {0} with description: {1} matches '
+ 'multiple tasks: {2}' .format(
task['uuid'] or task['id'], task['description'], output)
)
task['uuid'] or task['id'], task['description'], output)
)
projects / etc / taskwarrior.git / blobdiff