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:
from __future__ import print_function
import copy
from __future__ import print_function
import copy
import json
import logging
import os
import six
import sys
import json
import logging
import os
import six
import sys
-from backends import TaskWarrior, TaskWarriorException
-from serializing import SerializingObject
+from .serializing import SerializingObject
DATE_FORMAT = '%Y%m%dT%H%M%SZ'
REPR_OUTPUT_SIZE = 10
PENDING = 'pending'
COMPLETED = 'completed'
DATE_FORMAT = '%Y%m%dT%H%M%SZ'
REPR_OUTPUT_SIZE = 10
PENDING = 'pending'
COMPLETED = 'completed'
+DELETED = 'deleted'
+WAITING = 'waiting'
+RECURRING = 'recurring'
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
def __len__(self):
return len(self.viewed_dict)
def __len__(self):
return len(self.viewed_dict)
+ def __unicode__(self):
+ return six.u('ReadOnlyDictView: {0}'.format(repr(self.viewed_dict)))
+
+ __repr__ = __unicode__
+
def get(self, key, default=None):
return copy.deepcopy(self.viewed_dict.get(key, default))
def get(self, key, default=None):
return copy.deepcopy(self.viewed_dict.get(key, default))
if update_original:
self._original_data = copy.deepcopy(self._data)
if update_original:
self._original_data = copy.deepcopy(self._data)
def __getitem__(self, key):
# This is a workaround to make TaskResource non-iterable
# over simple index-based iteration
def __getitem__(self, key):
# This is a workaround to make TaskResource non-iterable
# over simple index-based iteration
# to pass that to TaskWarrior.
data_tuples = filter(lambda t: t[1] is not '', data_tuples)
data = dict(data_tuples)
# to pass that to TaskWarrior.
data_tuples = filter(lambda t: t[1] is not '', data_tuples)
data = dict(data_tuples)
- return json.dumps(data, separators=(',',':'))
+ return json.dumps(data, separators=(',', ':'))
@property
def _modified_fields(self):
@property
def _modified_fields(self):
def __init__(self, task, data=None):
self.task = task
self._load_data(data or dict())
def __init__(self, task, data=None):
self.task = task
self._load_data(data or dict())
- super(TaskAnnotation, self).__init__(task.warrior)
+ super(TaskAnnotation, self).__init__(task.backend)
def remove(self):
self.task.remove_annotation(self)
def remove(self):
self.task.remove_annotation(self)
# their data dics are the same
return self.task == other.task and self._data == other._data
# their data dics are the same
return self.task == other.task and self._data == other._data
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
- def from_input(cls, input_file=sys.stdin, modify=None, warrior=None):
+ def from_input(cls, input_file=sys.stdin, modify=None, backend=None):
"""
Creates a Task object, directly from the stdin, by reading one line.
If modify=True, two lines are used, first line interpreted as the
"""
Creates a Task object, directly from the stdin, by reading one line.
If modify=True, two lines are used, first line interpreted as the
modify = name.startswith('on-modify') if modify is None else modify
# Create the TaskWarrior instance if none passed
modify = name.startswith('on-modify') if modify is None else modify
# Create the TaskWarrior instance if none passed
+ if backend is None:
+ backends = importlib.import_module('tasklib.backends')
hook_parent_dir = os.path.dirname(os.path.dirname(sys.argv[0]))
hook_parent_dir = os.path.dirname(os.path.dirname(sys.argv[0]))
- warrior = TaskWarrior(data_location=hook_parent_dir)
+ backend = backends.TaskWarrior(data_location=hook_parent_dir)
# TaskWarrior instance is set to None
# TaskWarrior instance is set to None
# Load the data from the input
task._load_data(json.loads(input_file.readline().strip()))
# Load the data from the input
task._load_data(json.loads(input_file.readline().strip()))
- def __init__(self, warrior, **kwargs):
- super(Task, self).__init__(warrior)
+ def __init__(self, backend, **kwargs):
+ super(Task, self).__init__(backend)
# Check that user is not able to set read-only value in __init__
for key in kwargs.keys():
# Check that user is not able to set read-only value in __init__
for key in kwargs.keys():
# If the tasks are not saved, compare the actual instances
return id(self) == id(other)
# If the tasks are not saved, compare the actual instances
return id(self) == id(other)
+ def __ne__(self, other):
+ return not self.__eq__(other)
def __hash__(self):
if self['uuid']:
def __hash__(self):
if self['uuid']:
def pending(self):
return self['status'] == six.text_type('pending')
def pending(self):
return self['status'] == six.text_type('pending')
+ @property
+ def recurring(self):
+ return self['status'] == six.text_type('recurring')
+
@property
def active(self):
return self['start'] is not None
@property
def active(self):
return self['start'] is not None
# Check that all the tasks are saved
for task in (cur_dependencies or set()):
if not task.saved:
# Check that all the tasks are saved
for task in (cur_dependencies or set()):
if not task.saved:
- raise Task.NotSaved('Task \'%s\' needs to be saved before '
- 'it can be set as dependency.' % task)
+ raise Task.NotSaved(
+ 'Task \'%s\' needs to be saved before '
+ 'it can be set as dependency.' % task,
+ )
return super(Task, self).serialize_depends(cur_dependencies)
def delete(self):
if not self.saved:
return super(Task, self).serialize_depends(cur_dependencies)
def delete(self):
if not self.saved:
- raise Task.NotSaved("Task needs to be saved before it can be deleted")
+ raise Task.NotSaved(
+ 'Task needs to be saved before it can be deleted',
+ )
# Refresh the status, and raise exception if the task is deleted
self.refresh(only_fields=['status'])
if self.deleted:
# Refresh the status, and raise exception if the task is deleted
self.refresh(only_fields=['status'])
if self.deleted:
- raise Task.DeletedTask("Task was already deleted")
+ raise Task.DeletedTask('Task was already deleted')
self.backend.delete_task(self)
self.backend.delete_task(self)
def start(self):
if not self.saved:
def start(self):
if not self.saved:
- raise Task.NotSaved("Task needs to be saved before it can be started")
+ raise Task.NotSaved(
+ 'Task needs to be saved before it can be started',
+ )
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if self.completed:
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if self.completed:
- raise Task.CompletedTask("Cannot start a completed task")
+ raise Task.CompletedTask('Cannot start a completed task')
- raise Task.DeletedTask("Deleted task cannot be started")
+ raise Task.DeletedTask('Deleted task cannot be started')
- raise Task.ActiveTask("Task is already active")
+ raise Task.ActiveTask('Task is already active')
self.backend.start_task(self)
self.backend.start_task(self)
def stop(self):
if not self.saved:
def stop(self):
if not self.saved:
- raise Task.NotSaved("Task needs to be saved before it can be stopped")
+ raise Task.NotSaved(
+ 'Task needs to be saved before it can be stopped',
+ )
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if not self.active:
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if not self.active:
- raise Task.InactiveTask("Cannot stop an inactive task")
+ raise Task.InactiveTask('Cannot stop an inactive task')
self.backend.stop_task(self)
self.backend.stop_task(self)
def done(self):
if not self.saved:
def done(self):
if not self.saved:
- raise Task.NotSaved("Task needs to be saved before it can be completed")
+ raise Task.NotSaved(
+ 'Task needs to be saved before it can be completed',
+ )
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if self.completed:
# Refresh, and raise exception if task is already completed/deleted
self.refresh(only_fields=['status'])
if self.completed:
- raise Task.CompletedTask("Cannot complete a completed task")
+ raise Task.CompletedTask('Cannot complete a completed task')
- raise Task.DeletedTask("Deleted task cannot be completed")
+ raise Task.DeletedTask('Deleted task cannot be completed')
self.backend.complete_task(self)
self.backend.complete_task(self)
def add_annotation(self, annotation):
if not self.saved:
def add_annotation(self, annotation):
if not self.saved:
- raise Task.NotSaved("Task needs to be saved to add annotation")
+ raise Task.NotSaved('Task needs to be saved to add annotation')
self.backend.annotate_task(self, annotation)
self.refresh(only_fields=['annotations'])
def remove_annotation(self, annotation):
if not self.saved:
self.backend.annotate_task(self, annotation)
self.refresh(only_fields=['annotations'])
def remove_annotation(self, annotation):
if not self.saved:
- raise Task.NotSaved("Task needs to be saved to remove annotation")
+ raise Task.NotSaved('Task needs to be saved to remove annotation')
if isinstance(annotation, TaskAnnotation):
annotation = annotation['description']
if isinstance(annotation, TaskAnnotation):
annotation = annotation['description']
def refresh(self, only_fields=None, after_save=False):
# Raise error when trying to refresh a task that has not been saved
if not self.saved:
def refresh(self, only_fields=None, after_save=False):
# Raise error when trying to refresh a task that has not been saved
if not self.saved:
- raise Task.NotSaved("Task needs to be saved to be refreshed")
+ raise Task.NotSaved('Task needs to be saved to be refreshed')
new_data = self.backend.refresh_task(self, after_save=after_save)
if only_fields:
to_update = dict(
new_data = self.backend.refresh_task(self, after_save=after_save)
if only_fields:
to_update = dict(
- [(k, new_data.get(k)) for k in only_fields])
+ [(k, new_data.get(k)) for k in only_fields],
+ )
self._update_data(to_update, update_original=True)
else:
self._load_data(new_data)
self._update_data(to_update, update_original=True)
else:
self._load_data(new_data)
class TaskQuerySet(object):
"""
Represents a lazy lookup for a task objects.
"""
class TaskQuerySet(object):
"""
Represents a lazy lookup for a task objects.
"""
- def __init__(self, warrior=None, filter_obj=None):
- self.warrior = warrior
+ def __init__(self, backend, filter_obj=None):
+ self.backend = backend
self._result_cache = None
self._result_cache = None
- self.filter_obj = filter_obj or self.warrior.filter_class(warrior)
+ self.filter_obj = filter_obj or self.backend.filter_class(backend)
def __deepcopy__(self, memo):
"""
Deep copy of a QuerySet doesn't populate the cache
"""
def __deepcopy__(self, memo):
"""
Deep copy of a QuerySet doesn't populate the cache
"""
+ obj = self.__class__(backend=self.backend)
for k, v in self.__dict__.items():
if k in ('_iter', '_result_cache'):
obj.__dict__[k] = None
for k, v in self.__dict__.items():
if k in ('_iter', '_result_cache'):
obj.__dict__[k] = None
def __repr__(self):
data = list(self[:REPR_OUTPUT_SIZE + 1])
if len(data) > REPR_OUTPUT_SIZE:
def __repr__(self):
data = list(self[:REPR_OUTPUT_SIZE + 1])
if len(data) > REPR_OUTPUT_SIZE:
- data[-1] = "...(remaining elements truncated)..."
+ data[-1] = '...(remaining elements truncated)...'
return repr(data)
def __len__(self):
return repr(data)
def __len__(self):
if klass is None:
klass = self.__class__
filter_obj = self.filter_obj.clone()
if klass is None:
klass = self.__class__
filter_obj = self.filter_obj.clone()
- c = klass(warrior=self.warrior, filter_obj=filter_obj)
+ c = klass(backend=self.backend, filter_obj=filter_obj)
c.__dict__.update(kwargs)
return c
c.__dict__.update(kwargs)
return c
"""
Fetch the tasks which match the current filters.
"""
"""
Fetch the tasks which match the current filters.
"""
- return self.warrior.filter_tasks(self.filter_obj)
+ return self.backend.filter_tasks(self.filter_obj)
def completed(self):
return self.filter(status=COMPLETED)
def completed(self):
return self.filter(status=COMPLETED)
+ def deleted(self):
+ return self.filter(status=DELETED)
+
+ def waiting(self):
+ return self.filter(status=WAITING)
+
+ def recurring(self):
+ return self.filter(status=RECURRING)
+
def filter(self, *args, **kwargs):
"""
Returns a new TaskQuerySet with the given filters added.
def filter(self, *args, **kwargs):
"""
Returns a new TaskQuerySet with the given filters added.
if not num:
raise Task.DoesNotExist(
'Task matching query does not exist. '
if not num:
raise Task.DoesNotExist(
'Task matching query does not exist. '
- 'Lookup parameters were {0}'.format(kwargs))
+ 'Lookup parameters were {0}'.format(kwargs),
+ )
raise ValueError(
'get() returned more than one Task -- it returned {0}! '
raise ValueError(
'get() returned more than one Task -- it returned {0}! '
- 'Lookup parameters were {1}'.format(num, kwargs))
+ 'Lookup parameters were {1}'.format(num, kwargs),
+ )
projects / etc / taskwarrior.git / blobdiff