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)
+ 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._config = None
- self.version = self._get_version()
+ self.version = version_override or self._get_version()
self.overrides = {
'confirmation': 'no',
'dependency.confirmation': 'no', # See TW-1483 or taskrc man page
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)]
+ command_args = self._get_task_command()
overrides = self.overrides.copy()
overrides.update(config_override or dict())
for item in overrides.items():
def _get_version(self):
p = subprocess.Popen(
- ['task', '--version'],
+ self._get_task_command() + ['--version'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = [x.decode('utf-8') for x in p.communicate()]
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)
args, config_override=config_override)
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():