]>
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
from .filters import TaskWarriorFilter
from .task import Task, TaskQuerySet
from .filters import TaskWarriorFilter
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
class Backend(object):
@abc.abstractproperty
class Backend(object):
@abc.abstractproperty
-class TaskWarrior(object ):
+class TaskWarrior(Backend ):
VERSION_2_1_0 = six.u('2.1.0')
VERSION_2_2_0 = six.u('2.2.0')
VERSION_2_1_0 = six.u('2.1.0')
VERSION_2_2_0 = six.u('2.2.0')
if not os.path.exists(self.taskrc_location):
self.taskrc_location = '/'
if not os.path.exists(self.taskrc_location):
self.taskrc_location = '/'
self.version = self._get_version()
self.version = self._get_version()
'confirmation': 'no',
'dependency.confirmation': 'no', # See TW-1483 or taskrc man page
'recurrence.confirmation': 'no', # Necessary for modifying R tasks
'confirmation': 'no',
'dependency.confirmation': 'no', # See TW-1483 or taskrc man page
'recurrence.confirmation': 'no', # Necessary for modifying R tasks
data_location = os.path.expanduser(data_location)
if create and not os.path.exists(data_location):
os.makedirs(data_location)
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_command_args(self, args, config_override=None):
command_args = ['task', 'rc:{0}'.format(self.taskrc_location)]
self.tasks = TaskQuerySet(self)
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():
+ 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))
return command_args
def _get_version(self):
p = subprocess.Popen(
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)
+ ['task', '--version'],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
return stdout.strip('\n')
stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
return stdout.strip('\n')
if serialized_value is '':
escaped_serialized_value = ''
else:
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)
format_func = getattr(self, 'format_{0}'.format(field),
format_default)
# Removed dependencies need to be prefixed with '-'
return 'depends:' + ','.join(
# 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
def format_description(self, task):
# Task version older than 2.4.0 ignores first word of the
+ @property
+ def config(self):
+ # First, check if memoized information is available
+ if self._config:
+ return copy.deepcopy(self._config)
+
+ # If not, fetch the config using the 'show' command
raw_output = self.execute_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 = dict()
config_regex = re.compile(r'^(?P<key>[^\s]+)\s+(?P<value>[^\s].+$)')
if match:
config[match.group('key')] = match.group('value').strip()
if match:
config[match.group('key')] = match.group('value').strip()
+ # Memoize the config dict
+ self._config = config
+
+ return copy.deepcopy(config)
def execute_command(self, args, config_override=None, allow_failure=True,
return_all=False):
def execute_command(self, args, config_override=None, allow_failure=True,
return_all=False):
taskfilter.add_filter_param(key, value)
output = self.execute_command(['export', '--'] +
taskfilter.add_filter_param(key, value)
output = self.execute_command(['export', '--'] +
- taskfilter.get_filter_params())
+ 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(
# 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)
+ task['uuid'] or task['id'], task['description'], output)
)
return json.loads(output[0])
)
return json.loads(output[0])