from logging import getLogger
from typing import Optional, List, Awaitable
from nextcloud_tasks_api import NextcloudTasksApi, TaskList
from nextcloud_tasks_api.ical import Task
from textual.app import ComposeResult
from textual.containers import Vertical, Horizontal, Grid
from textual.widgets import Button
from .dialog import CreateDialog, CreateAnswer, DeleteDialog, CreateListDialog, DeleteListDialog
from .entry import TaskEntry
from .repo import TaskUpdater, TaskRef
from .status import StatusBar
from .tree import TaskTree, TaskTreeNodeTree, TaskTreeNode, TaskListSelect
class TaskWidget(Horizontal):
DEFAULT_CSS = """
TaskWidget TaskEntry {
padding: 1;
}
#tree-wrap {
padding: 1;
}
#list-box {
height: auto;
padding: 1 1 0 0;
grid-size: 4 1;
grid-columns: 1fr 5 5 5;
grid-gutter: 1;
}
SelectCurrent {
border: wide $accent;
}
#list-box Button {
width: 5;
min-width: 5;
}
#button-box {
padding: 0 1 1 1;
height: auto;
grid-size: 3 1;
grid-gutter: 1;
}
#button-box Button {
width: 1fr;
}
"""
def __init__(self, api: NextcloudTasksApi, status: StatusBar, default_list: Optional[str]) -> None:
super().__init__()
self._status: StatusBar = status
self._repo: TaskUpdater = TaskUpdater(api, self._status, self._updated)
self._default_list: Optional[str] = default_list
self._select: TaskListSelect = TaskListSelect([], prompt="Choose List", allow_blank=True)
self._tree: TaskTree = TaskTree()
self._entry: TaskEntry = TaskEntry(status, self._repo)
self._btn_create: Button = Button(label="Create", variant="primary", id="button-create")
self._btn_delete: Button = Button(label="Delete", variant="error", id="button-delete")
self._btn_refresh: Button = Button(label="Refresh", variant="success", id="button-refresh")
self._btn_create_list: Button = Button(label="+", variant="primary", id="button-create-list")
self._btn_delete_list: Button = Button(label="-", variant="error", id="button-delete-list")
self._btn_refresh_list: Button = Button(label="⇅", variant="success", id="button-refresh-list")
def _updated(self, task: TaskRef) -> None:
self._tree.update_node(TaskTreeNodeTree.from_task(task.ical))
def compose(self) -> ComposeResult:
yield Vertical(Grid(self._select, self._btn_create_list,
self._btn_delete_list, self._btn_refresh_list, id="list-box"),
Vertical(self._tree, id="tree-wrap"),
Grid(self._btn_create, self._btn_delete, self._btn_refresh, id="button-box"))
yield self._entry
def on_mount(self) -> None:
self._run_busy_worker(self._do_get_lists(self._default_list))
async def on_unmount(self) -> None:
await self._repo.reset(None)
def on_button_pressed(self, event: Button.Pressed) -> None:
cursor_uid: Optional[str] = self._tree.cursor_uid
if event.button.id == "button-create":
self._run_modal_worker(self._do_create_task(cursor_uid))
elif event.button.id == "button-delete":
if cursor_uid is not None and self._entry.uid is not None and cursor_uid == self._entry.uid:
self._run_modal_worker(self._do_delete_task(cursor_uid))
elif event.button.id == "button-refresh":
self._run_busy_worker(self._do_list_tasks(self._repo.current_list))
elif event.button.id == "button-create-list":
self._run_busy_worker(self._do_create_list())
elif event.button.id == "button-delete-list":
self._run_busy_worker(self._do_delete_list())
elif event.button.id == "button-refresh-list":
self._run_busy_worker(self._do_get_lists())
def _update_buttons(self) -> None:
# TODO: can somehow unbind, hide, or disable key bindings?
self._btn_delete_list.disabled = self._repo.current_list is None
self._btn_create.disabled = self._repo.current_list is None
self._btn_delete.disabled = self._entry.uid is None
self._btn_refresh.disabled = self._repo.current_list is None
def on_task_list_select_updated(self, evt: TaskListSelect.Updated) -> None:
if evt.task_list is not self._repo.current_list:
self._run_busy_worker(self._do_list_tasks(evt.task_list))
def on_task_tree_highlight_action(self, evt: TaskTree.HighlightAction) -> None:
self._entry.uid = evt.uid
self._update_buttons()
def on_task_tree_toggle_action(self, evt: TaskTree.ToggleAction) -> None:
if evt.uid is not None and self._entry.uid is not None and evt.uid == self._entry.uid:
self._entry.toggle(evt.uid)
def on_task_tree_select_action(self, evt: TaskTree.SelectAction) -> None:
if evt.uid is not None and self._entry.uid is not None and evt.uid == self._entry.uid:
self._entry.set_focus()
def on_task_tree_create_action(self, evt: TaskTree.CreateAction) -> None:
if evt.uid is not None and self._entry.uid is not None and evt.uid == self._entry.uid:
self._run_modal_worker(self._do_create_task(evt.uid))
elif evt.uid is None and self._entry.uid is None:
self._run_modal_worker(self._do_create_task())
def on_task_tree_delete_action(self, evt: TaskTree.DeleteAction) -> None:
if evt.uid is not None and self._entry.uid is not None and evt.uid == self._entry.uid:
self._run_modal_worker(self._do_delete_task(evt.uid))
def on_task_tree_quit_action(self, evt: TaskTree.QuitAction) -> None:
self._select.focus()
def on_task_entry_quit_action(self, evt: TaskEntry.QuitAction) -> None:
self._tree.focus()
async def _do_get_lists(self, preselect_list: Optional[str] = None) -> None:
self._select.update([_ async for _ in self._repo.get_lists()], preselect_list)
await self._do_list_tasks(self._select.task_list) # waiting for select update worker could cause flicker
async def _do_delete_list(self) -> None:
current_list: Optional[TaskList] = self._repo.current_list
if current_list is not None:
answer: Optional[bool] = await self.app.push_screen_wait(DeleteListDialog(current_list.name or ""))
if answer is True:
await self._repo.join()
await self._repo.delete_current_list(current_list)
getLogger("Deleted task list").warning(current_list.name or "")
await self._do_get_lists()
async def _do_create_list(self) -> None:
answer: Optional[str] = await self.app.push_screen_wait(CreateListDialog())
if answer is not None:
task_list: Optional[TaskList] = await self._repo.create_list(answer)
if task_list is not None:
getLogger("Created task list").info(answer)
await self._do_get_lists(task_list.name)
async def _do_list_tasks(self, task_list: Optional[TaskList]) -> None:
self._entry.uid = None
await self._repo.reset(task_list)
try:
self._tree.update(TaskTreeNodeTree.build([_.ical for _ in await self._repo.list_tasks()]))
except (KeyError, ValueError) as e:
getLogger("Load list").error(str(e))
await self._repo.reset(None)
self._tree.update([])
task_list = None
if task_list is not None:
self._tree.focus()
else:
self._select.focus()
self._select.expanded = True
async def _do_create_task(self, parent_uid: Optional[str] = None) -> None:
parent_task: Optional[Task] = await self._repo.get(parent_uid) if parent_uid is not None else None
answer: Optional[CreateAnswer] = await self.app.push_screen_wait(
CreateDialog(parent_task.uid if parent_task is not None else None,
parent_task.summary if parent_task is not None else None)
)
if answer is not None:
async for task in self._repo.create_task(parent_task.uid if answer.parent and parent_task else None,
answer.summary):
try:
self._tree.add_node(TaskTreeNodeTree.from_task(task))
except (ValueError, KeyError) as e:
getLogger("Create task").error(str(e))
if answer.parent and parent_task:
getLogger("Created subtask").info("\n".join(answer.summary))
else:
getLogger("Created task").info("\n".join(answer.summary))
self._tree.focus()
async def _do_delete_task(self, parent_uid: str) -> None:
await self._repo.join()
parent: Task = await self._repo.get(parent_uid)
children: List[List[TaskTreeNode]] = list(self._tree.dfs_tree(self._tree.get_node(parent_uid)))
answer: Optional[bool] = await self.app.push_screen_wait(
DeleteDialog(parent.summary or "", sum(len(_) for _ in children))
)
if answer is True:
await self._repo.delete_task([[__.uid for __ in _] for _ in children])
self._tree.remove_nodes([__ for _ in children for __ in _])
getLogger("Deleted task").warning(parent.summary or "")
self._tree.focus()
def _run_modal_worker(self, work: Awaitable[None]) -> bool:
async def wrapper() -> None:
self._entry.flush()
await work
self._update_buttons()
self.disabled = False
if not self.disabled:
self.disabled = True
self.run_worker(wrapper())
return True
else:
return False
def _run_busy_worker(self, work: Awaitable[None]) -> None:
async def wrapper() -> None:
await work
self._status.busy -= 1
if self._run_modal_worker(wrapper()):
self._status.busy += 1
