15654091449

Committed 14 Jun 2025 04:40PM UTC coverage: 94.09% (-0.07%) from 94.157%

Build # 15654091449

Build Type

push

github

Committed by

tbabej

Commit Message

Adds expiration date to viewports and preset headers

Coverage Stats

40 of 43 new or added lines in 5 files covered. (93.02%)

1592 of 1692 relevant lines covered (94.09%)

14.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.95

/taskwiki/cache.py

import vim  # pylint: disable=F0401
import re
import six

from taskwiki import preset
from taskwiki import viewport
from taskwiki import regexp
from taskwiki import store
from taskwiki import short
from taskwiki import util
from taskwiki import errors


class BufferProxy(object):

    def __init__(self, number):
        self.data = []
        self.buffer_number = number

    def obtain(self):
        self.data = util.get_buffer(self.buffer_number)[:]

    def push(self):
        with util.current_line_preserved():
            buffer = util.get_buffer(self.buffer_number)
            # Only set the buffer contents if the data is changed.
            # Avoids extra undo events with empty diff.
            if buffer[:] != self.data:
                buffer[:] = self.data
                buffer.options['modified'] = True

    def __getitem__(self, index):
        try:
            return self.data[index]
        except IndexError:
            return ''

    def __setitem__(self, index, lines):
        self.data[index] = lines

    def __delitem__(self, index):
        del self.data[index]

    def __iter__(self):
        for line in self.data:
            yield line

    def __len__(self):
        return len(self.data)

    def append(self, data, position=None):
        if position is None:
            self.data.append(data)
        else:
            self.data.insert(position, data)


class CacheRegistry(object):
    """
    Provide a registry for TaskCache instances related to vim buffers.
    """

    def __init__(self):
        # Store caches indexed by buffer number
        self.caches = {}

        # Remember current cache
        self.current_buffer = None

    def __call__(self, buffer_number = None):
        """
        Get cache for given buffer_number or the one which was accessed most recently.
        """

        if buffer_number is None:
            buffer_number = self.current_buffer
        else:
            self.current_buffer = buffer_number

        try:
            # Use existing cache if it was loaded before
            cache = self.caches[buffer_number]
        except KeyError:
            cache = self._load_cache(buffer_number)

        return cache

    def _load_cache(self, buffer_number):
        # Initialize the cache
        cache = TaskCache(buffer_number)
        self.caches[buffer_number] = cache

        # Check the necessary dependencies
        util.enforce_dependencies(cache)

        return cache

    def load_current(self):
        return self(vim.current.buffer.number)


class TaskCache(object):
    """
    A cache that holds all the tasks in the given buffer and prevents
    multiple redundant taskwarrior calls.
    """

    def __init__(self, buffer_number):
        # Determine defaults
        default_rc = util.get_var('taskwiki_taskrc_location') or '~/.taskrc'
        default_data = util.get_var('taskwiki_data_location') or None
        extra_warrior_defs = util.get_var('taskwiki_extra_warriors', {})
        markup_syntax = vim.eval("vimwiki#vars#get_wikilocal('syntax')") or 'default'

        # Validate markup choice and set it
        if markup_syntax in ["default", "markdown"]:
            self.markup_syntax = markup_syntax
        else:
            msg = "Unknown markup given: {}".format(markup_syntax)
            raise errors.TaskWikiException(msg)

        # Initialize all the subcomponents
        self.buffer = BufferProxy(buffer_number)
        self.completion = store.CompletionStore(self)
        self.task = store.TaskStore(self)
        self.presets = store.PresetStore(self)
        self.vwtask = store.VwtaskStore(self)
        self.viewport = store.ViewportStore(self)
        self.line = store.LineStore(self)
        self.warriors = store.WarriorStore(default_rc, default_data, extra_warrior_defs)
        self.buffer_has_authority = True

    @property
    def vimwikitask_dependency_order(self):
        iterated_cache = {
            k:v for k,v in self.vwtask.items()
            if v is not None
        }

        while iterated_cache.keys():
            for key in list(iterated_cache.keys()):
                task = iterated_cache[key]
                if all([t['line_number'] not in iterated_cache.keys()
                        for t in task.add_dependencies]):
                    del iterated_cache[key]
                    yield task

    def reset(self):
        self.buffer.obtain()
        self.completion.store = dict()
        self.task.store = dict()
        self.presets.store = dict()
        self.vwtask.store = dict()
        self.viewport.store = dict()
        self.line.store = dict()

    def load_presets(self):
        stack = []

        for i in range(len(self.buffer)):
            header = preset.PresetHeader.from_line(i, self, stack[-1] if stack else None)

            if header is None:
                continue

            # Save the preset header in the cache
            self.presets[i] = header

            while stack and stack[-1].level >= header.level:
                stack.pop()

            stack.append(header)

    def load_vwtasks(self, buffer_has_authority=True):
        # Set the authority flag, which determines which data (Buffer or TW)
        # will be considered authoritative
        old_authority = self.buffer_has_authority
        self.buffer_has_authority = buffer_has_authority

        for i in range(len(self.buffer)):
            self.vwtask[i]  # Loads the line into the cache

        # Restore the old authority flag value
        self.buffer_has_authority = old_authority

    def load_viewports(self):
        for i in range(len(self.buffer)):
            port = viewport.ViewPort.from_line(i, self)

            if port is None:
                continue

            port.load_tasks()

            # Save the viewport in the cache
            self.viewport[i] = port

    def update_vwtasks_in_buffer(self):
        for vwtask in self.vwtask.values():
            port = self.get_viewport_by_task(vwtask.task)
            if port and port.expired:
                continue
            vwtask.update_in_buffer()

    def save_tasks(self):
        for vwtask in self.vimwikitask_dependency_order:
            port = self.get_viewport_by_task(vwtask.task)
            if port and port.expired:
                continue
            vwtask.save_to_tw()

    def load_tasks(self):
        raw_task_info = []

        # Load the tasks in batch, all in given TaskWarrior instance
        for line in self.buffer:
            match = re.search(regexp.GENERIC_TASK, line)
            if not match:
                continue

            tw = self.warriors[match.group('source') or 'default']
            uuid = match.group('uuid')

            if not uuid:
                continue

            raw_task_info.append((uuid, tw))

        for tw in self.warriors.values():
            # Select all tasks in the files that have UUIDs
            uuids = [uuid for uuid, task_tw in raw_task_info if task_tw == tw]

            # If no task in the file contains UUID, we have no job here
            if not uuids:
                continue

            # Get them out of TaskWarrior at once
            tasks = tw.tasks.filter()
            for uuid in uuids:
                tasks = tasks.filter(uuid=uuid)

            # Update each task in the cache
            for task in tasks:
                key = short.ShortUUID(task['uuid'], tw)
                self.task[key] = task

    def update_vwtasks_from_tasks(self):
        for vwtask in self.vwtask.values():
            port = self.get_viewport_by_task(vwtask.task)
            if port and port.expired:
                continue
            vwtask.update_from_task()

    def evaluate_viewports(self):
        for port in self.viewport.values():
            if port.expired:
                continue
            port.sync_with_taskwarrior()

    def get_viewport_by_task(self, task):
        """
        Looks for a viewport containing the given task by iterating over the cached
        items.

        Returns the viewport, or None if not found.
        """

        for port in self.viewport.values():
            if task in port.viewport_tasks:
                return port

    def insert_line(self, line, position):
        # Insert the line
        if position == len(self.buffer):
            # Workaround: Necessary since neovim cannot append
            # after the last line of the buffer when mentioning
            # the position explicitly
            self.buffer.append(line)
        else:
            self.buffer.append(line, position)

        # Update the position of all the things shifted by the insertion
        self.vwtask.shift(position, 1)
        self.viewport.shift(position, 1)

        # Shift lines in the line cache
        self.line.shift(position, 1)

    def remove_line(self, position):
        # Remove the line
        del self.buffer[position]

        # Remove the vimwikitask from cache
        del self.vwtask[position]
        del self.viewport[position]

        # Delete the line from the line cache
        del self.line[position]

        # Update the position of all the things shifted by the removal
        self.vwtask.shift(position, -1)
        self.viewport.shift(position, -1)

        # Shift lines in the line cache
        self.line.shift(position, -1)

    def swap_lines(self, position1, position2):
        buffer_size = len(self.buffer)
        if position1 >= buffer_size or position2 >= buffer_size:
            raise ValueError("Tring to swap %d with %d" % (position1, position2))

        # Swap lines in the line cache
        self.line.swap(position1, position2)

        # Swap both the viewport and vimwikitasks indexes
        self.vwtask.swap(position1, position2)
        self.viewport.swap(position1, position2)

    def get_relevant_tw(self):
        # Find closest task
        from taskwiki import vwtask
        task = vwtask.VimwikiTask.find_closest(self)
        return task.tw if task else self.warriors['default']

    def get_relevant_completion(self):
        return self.completion[self.get_relevant_tw()]

1	import vim # pylint: disable=F0401	15✔
2	import re	15✔
3	import six	15✔
4
5	from taskwiki import preset	15✔
6	from taskwiki import viewport	15✔
7	from taskwiki import regexp	15✔
8	from taskwiki import store	15✔
9	from taskwiki import short	15✔
10	from taskwiki import util	15✔
11	from taskwiki import errors	15✔
12
13
14	class BufferProxy(object):	15✔
15
16	def __init__(self, number):	15✔
17	self.data = []	15✔
18	self.buffer_number = number	15✔
19
20	def obtain(self):	15✔
21	self.data = util.get_buffer(self.buffer_number)[:]	15✔
22
23	def push(self):	15✔
24	with util.current_line_preserved():	15✔
25	buffer = util.get_buffer(self.buffer_number)	15✔
26	# Only set the buffer contents if the data is changed.
27	# Avoids extra undo events with empty diff.
28	if buffer[:] != self.data:	15✔
29	buffer[:] = self.data	15✔
30	buffer.options['modified'] = True	15✔
31
32	def __getitem__(self, index):	15✔
33	try:	15✔
34	return self.data[index]	15✔
35	except IndexError:	15✔
36	return ''	15✔
37
38	def __setitem__(self, index, lines):	15✔
39	self.data[index] = lines	15✔
40
41	def __delitem__(self, index):	15✔
42	del self.data[index]	15✔
43
44	def __iter__(self):	15✔
45	for line in self.data:	15✔
46	yield line	15✔
47
48	def __len__(self):	15✔
49	return len(self.data)	15✔
50
51	def append(self, data, position=None):	15✔
52	if position is None:	15✔
53	self.data.append(data)	15✔
54	else:
55	self.data.insert(position, data)	15✔
56
57
58	class CacheRegistry(object):	15✔
59	"""
60	Provide a registry for TaskCache instances related to vim buffers.
61	"""
62
63	def __init__(self):	15✔
64	# Store caches indexed by buffer number
65	self.caches = {}	15✔
66
67	# Remember current cache
68	self.current_buffer = None	15✔
69
70	def __call__(self, buffer_number = None):	15✔
71	"""
72	Get cache for given buffer_number or the one which was accessed most recently.
73	"""
74
75	if buffer_number is None:	15✔
76	buffer_number = self.current_buffer	15✔
77	else:
78	self.current_buffer = buffer_number	15✔
79
80	try:	15✔
81	# Use existing cache if it was loaded before
82	cache = self.caches[buffer_number]	15✔
83	except KeyError:	15✔
84	cache = self._load_cache(buffer_number)	15✔
85
86	return cache	15✔
87
88	def _load_cache(self, buffer_number):	15✔
89	# Initialize the cache
90	cache = TaskCache(buffer_number)	15✔
91	self.caches[buffer_number] = cache	15✔
92
93	# Check the necessary dependencies
94	util.enforce_dependencies(cache)	15✔
95
96	return cache	15✔
97
98	def load_current(self):	15✔
99	return self(vim.current.buffer.number)	15✔
100
101
102	class TaskCache(object):	15✔
103	"""
104	A cache that holds all the tasks in the given buffer and prevents
105	multiple redundant taskwarrior calls.
106	"""
107
108	def __init__(self, buffer_number):	15✔
109	# Determine defaults
110	default_rc = util.get_var('taskwiki_taskrc_location') or '~/.taskrc'	15✔
111	default_data = util.get_var('taskwiki_data_location') or None	15✔
112	extra_warrior_defs = util.get_var('taskwiki_extra_warriors', {})	15✔
113	markup_syntax = vim.eval("vimwiki#vars#get_wikilocal('syntax')") or 'default'	15✔
114
115	# Validate markup choice and set it
116	if markup_syntax in ["default", "markdown"]:	15✔
117	self.markup_syntax = markup_syntax	15✔
118	else:
119	msg = "Unknown markup given: {}".format(markup_syntax)	×
120	raise errors.TaskWikiException(msg)	×
121
122	# Initialize all the subcomponents
123	self.buffer = BufferProxy(buffer_number)	15✔
124	self.completion = store.CompletionStore(self)	15✔
125	self.task = store.TaskStore(self)	15✔
126	self.presets = store.PresetStore(self)	15✔
127	self.vwtask = store.VwtaskStore(self)	15✔
128	self.viewport = store.ViewportStore(self)	15✔
129	self.line = store.LineStore(self)	15✔
130	self.warriors = store.WarriorStore(default_rc, default_data, extra_warrior_defs)	15✔
131	self.buffer_has_authority = True	15✔
132
133	@property	15✔
134	def vimwikitask_dependency_order(self):	15✔
135	iterated_cache = {	15✔
136	k:v for k,v in self.vwtask.items()
137	if v is not None
138	}
139
140	while iterated_cache.keys():	15✔
141	for key in list(iterated_cache.keys()):	15✔
142	task = iterated_cache[key]	15✔
143	if all([t['line_number'] not in iterated_cache.keys()	15✔
144	for t in task.add_dependencies]):
145	del iterated_cache[key]	15✔
146	yield task	15✔
147
148	def reset(self):	15✔
149	self.buffer.obtain()	15✔
150	self.completion.store = dict()	15✔
151	self.task.store = dict()	15✔
152	self.presets.store = dict()	15✔
153	self.vwtask.store = dict()	15✔
154	self.viewport.store = dict()	15✔
155	self.line.store = dict()	15✔
156
157	def load_presets(self):	15✔
158	stack = []	15✔
159
160	for i in range(len(self.buffer)):	15✔
161	header = preset.PresetHeader.from_line(i, self, stack[-1] if stack else None)	15✔
162
163	if header is None:	15✔
164	continue	15✔
165
166	# Save the preset header in the cache
167	self.presets[i] = header	15✔
168
169	while stack and stack[-1].level >= header.level:	15✔
170	stack.pop()	15✔
171
172	stack.append(header)	15✔
173
174	def load_vwtasks(self, buffer_has_authority=True):	15✔
175	# Set the authority flag, which determines which data (Buffer or TW)
176	# will be considered authoritative
177	old_authority = self.buffer_has_authority	15✔
178	self.buffer_has_authority = buffer_has_authority	15✔
179
180	for i in range(len(self.buffer)):	15✔
181	self.vwtask[i] # Loads the line into the cache	15✔
182
183	# Restore the old authority flag value
184	self.buffer_has_authority = old_authority	15✔
185
186	def load_viewports(self):	15✔
187	for i in range(len(self.buffer)):	15✔
188	port = viewport.ViewPort.from_line(i, self)	15✔
189
190	if port is None:	15✔
191	continue	15✔
192
193	port.load_tasks()	15✔
194
195	# Save the viewport in the cache
196	self.viewport[i] = port	15✔
197
198	def update_vwtasks_in_buffer(self):	15✔
199	for vwtask in self.vwtask.values():	15✔
200	port = self.get_viewport_by_task(vwtask.task)	15✔
201	if port and port.expired:	15✔
202	continue	15✔
203	vwtask.update_in_buffer()	15✔
204
205	def save_tasks(self):	15✔
206	for vwtask in self.vimwikitask_dependency_order:	15✔
207	port = self.get_viewport_by_task(vwtask.task)	15✔
208	if port and port.expired:	15✔
209	continue	15✔
210	vwtask.save_to_tw()	15✔
211
212	def load_tasks(self):	15✔
213	raw_task_info = []	15✔
214
215	# Load the tasks in batch, all in given TaskWarrior instance
216	for line in self.buffer:	15✔
217	match = re.search(regexp.GENERIC_TASK, line)	15✔
218	if not match:	15✔
219	continue	15✔
220
221	tw = self.warriors[match.group('source') or 'default']	15✔
222	uuid = match.group('uuid')	15✔
223
224	if not uuid:	15✔
225	continue	15✔
226
227	raw_task_info.append((uuid, tw))	15✔
228
229	for tw in self.warriors.values():	15✔
230	# Select all tasks in the files that have UUIDs
231	uuids = [uuid for uuid, task_tw in raw_task_info if task_tw == tw]	15✔
232
233	# If no task in the file contains UUID, we have no job here
234	if not uuids:	15✔
235	continue	15✔
236
237	# Get them out of TaskWarrior at once
238	tasks = tw.tasks.filter()	15✔
239	for uuid in uuids:	15✔
240	tasks = tasks.filter(uuid=uuid)	15✔
241
242	# Update each task in the cache
243	for task in tasks:	15✔
244	key = short.ShortUUID(task['uuid'], tw)	15✔
245	self.task[key] = task	15✔
246
247	def update_vwtasks_from_tasks(self):	15✔
248	for vwtask in self.vwtask.values():	15✔
249	port = self.get_viewport_by_task(vwtask.task)	15✔
250	if port and port.expired:	15✔
NEW 251	continue	×
252	vwtask.update_from_task()	15✔
253
254	def evaluate_viewports(self):	15✔
255	for port in self.viewport.values():	15✔
256	if port.expired:	15✔
257	continue	15✔
258	port.sync_with_taskwarrior()	15✔
259
260	def get_viewport_by_task(self, task):	15✔
261	"""
262	Looks for a viewport containing the given task by iterating over the cached
263	items.
264
265	Returns the viewport, or None if not found.
266	"""
267
268	for port in self.viewport.values():	15✔
269	if task in port.viewport_tasks:	15✔
270	return port	15✔
271
272	def insert_line(self, line, position):	15✔
273	# Insert the line
274	if position == len(self.buffer):	15✔
275	# Workaround: Necessary since neovim cannot append
276	# after the last line of the buffer when mentioning
277	# the position explicitly
278	self.buffer.append(line)	15✔
279	else:
280	self.buffer.append(line, position)	15✔
281
282	# Update the position of all the things shifted by the insertion
283	self.vwtask.shift(position, 1)	15✔
284	self.viewport.shift(position, 1)	15✔
285
286	# Shift lines in the line cache
287	self.line.shift(position, 1)	15✔
288
289	def remove_line(self, position):	15✔
290	# Remove the line
291	del self.buffer[position]	15✔
292
293	# Remove the vimwikitask from cache
294	del self.vwtask[position]	15✔
295	del self.viewport[position]	15✔
296
297	# Delete the line from the line cache
298	del self.line[position]	15✔
299
300	# Update the position of all the things shifted by the removal
301	self.vwtask.shift(position, -1)	15✔
302	self.viewport.shift(position, -1)	15✔
303
304	# Shift lines in the line cache
305	self.line.shift(position, -1)	15✔
306
307	def swap_lines(self, position1, position2):	15✔
308	buffer_size = len(self.buffer)	15✔
309	if position1 >= buffer_size or position2 >= buffer_size:	15✔
310	raise ValueError("Tring to swap %d with %d" % (position1, position2))	×
311
312	# Swap lines in the line cache
313	self.line.swap(position1, position2)	15✔
314
315	# Swap both the viewport and vimwikitasks indexes
316	self.vwtask.swap(position1, position2)	15✔
317	self.viewport.swap(position1, position2)	15✔
318
319	def get_relevant_tw(self):	15✔
320	# Find closest task
321	from taskwiki import vwtask	15✔
322	task = vwtask.VimwikiTask.find_closest(self)	15✔
323	return task.tw if task else self.warriors['default']	15✔
324
325	def get_relevant_completion(self):	15✔
326	return self.completion[self.get_relevant_tw()]	15✔

tbabej / taskwiki / 15654091449

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous