From 859fefeb5362723752077e253ee431c0d1087a13 Mon Sep 17 00:00:00 2001 From: premchand11 Date: Wed, 1 Oct 2025 16:23:33 +0530 Subject: [PATCH] feat(storage): add indexing, pagination, and lazy loading to JSON TaskStorage --- tests/test_storage.py | 43 ++++++++++++++++++++ tix/storage/json_storage.py | 79 +++++++++++++++++++++++-------------- 2 files changed, 93 insertions(+), 29 deletions(-) diff --git a/tests/test_storage.py b/tests/test_storage.py index 1f91904..a1bef17 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -61,6 +61,8 @@ def test_backward_compatibility(temp_storage): """Test reading old format (plain list) and upgrading""" old_data = [{"id": 5, "text": "legacy", "priority": "low", "tags": [], "completed": False}] temp_storage.storage_path.write_text(json.dumps(old_data)) + # Force reload after overwriting file manually + temp_storage._load_index() tasks = temp_storage.load_tasks() assert len(tasks) == 1 @@ -74,3 +76,44 @@ def test_backward_compatibility(temp_storage): assert isinstance(data, dict) assert "next_id" in data assert "tasks" in data + + + +# --- new tests below --- + +def test_pagination(temp_storage): + """Test pagination with list_tasks""" + for i in range(25): + temp_storage.add_task(f"Task {i}") + + page1 = temp_storage.list_tasks(page=1, page_size=10) + page2 = temp_storage.list_tasks(page=2, page_size=10) + + assert len(page1) == 10 + assert len(page2) == 10 + assert page1[0].text == "Task 0" + assert page2[0].text == "Task 10" + + +def test_lazy_iteration(temp_storage): + """Test lazy iteration with iter_tasks""" + for i in range(5): + temp_storage.add_task(f"Lazy {i}") + + tasks = list(temp_storage.iter_tasks(0, 3)) + assert len(tasks) == 3 + assert tasks[0].text.startswith("Lazy") + + +def test_active_and_completed(temp_storage): + """Test filtering active and completed tasks""" + t1 = temp_storage.add_task("Active task") + t2 = temp_storage.add_task("Completed task") + t2.mark_done() + temp_storage.update_task(t2) + + active = temp_storage.get_active_tasks() + completed = temp_storage.get_completed_tasks() + + assert any(t.text == "Active task" for t in active) + assert any(t.text == "Completed task" for t in completed) diff --git a/tix/storage/json_storage.py b/tix/storage/json_storage.py index 53a8b8b..37110c2 100644 --- a/tix/storage/json_storage.py +++ b/tix/storage/json_storage.py @@ -5,7 +5,7 @@ class TaskStorage: - """JSON-based storage for tasks""" + """JSON-based storage for tasks (with indexing, pagination, and lazy loading)""" def __init__(self, storage_path: Path = None): """Initialize storage with default or custom path""" @@ -13,6 +13,11 @@ def __init__(self, storage_path: Path = None): self.storage_path.parent.mkdir(parents=True, exist_ok=True) self._ensure_file() + # --- new: build in-memory index for fast lookups --- + self._index: dict[int, Task] = {} + self._next_id: int = 1 + self._load_index() + def _ensure_file(self): """Ensure storage file exists""" if not self.storage_path.exists(): @@ -52,60 +57,76 @@ def _read_data(self) -> dict: return {"next_id": 1, "tasks": []} def _write_data(self, data: dict): + """Write raw data to storage""" self.storage_path.write_text(json.dumps(data, indent=2)) + def _load_index(self): + """Load tasks into memory and build index""" + data = self._read_data() + self._index = {t["id"]: Task.from_dict(t) for t in data["tasks"]} + self._next_id = data["next_id"] + + def _save_index(self): + """Persist index back to storage""" + data = { + "next_id": self._next_id, + "tasks": [t.to_dict() for t in self._index.values()], + } + self._write_data(data) + def load_tasks(self) -> List[Task]: """Load all tasks from storage""" - data = self._read_data() - return [Task.from_dict(item) for item in data["tasks"]] + return list(self._index.values()) def save_tasks(self, tasks: List[Task]): """Save all tasks to storage""" - data = self._read_data() - data["tasks"] = [task.to_dict() for task in tasks] - self._write_data(data) + self._index = {t.id: t for t in tasks} + self._next_id = max((t.id for t in tasks), default=0) + 1 + self._save_index() def add_task(self, text: str, priority: str = 'medium', tags: List[str] = None) -> Task: """Add a new task and return it""" - data = self._read_data() - new_id = data["next_id"] + new_id = self._next_id new_task = Task(id=new_id, text=text, priority=priority, tags=tags or []) - data["tasks"].append(new_task.to_dict()) - data["next_id"] = new_id + 1 - self._write_data(data) + self._index[new_id] = new_task + self._next_id += 1 + self._save_index() return new_task def get_task(self, task_id: int) -> Optional[Task]: """Get a specific task by ID""" - tasks = self.load_tasks() - for task in tasks: - if task.id == task_id: - return task - return None + return self._index.get(task_id) def update_task(self, task: Task): """Update an existing task""" - tasks = self.load_tasks() - for i, t in enumerate(tasks): - if t.id == task.id: - tasks[i] = task - self.save_tasks(tasks) - return + if task.id in self._index: + self._index[task.id] = task + self._save_index() def delete_task(self, task_id: int) -> bool: """Delete a task by ID, return True if deleted""" - tasks = self.load_tasks() - original_count = len(tasks) - new_tasks = [t for t in tasks if t.id != task_id] - if len(new_tasks) < original_count: - self.save_tasks(new_tasks) + if task_id in self._index: + del self._index[task_id] + self._save_index() return True return False def get_active_tasks(self) -> List[Task]: """Get all incomplete tasks""" - return [t for t in self.load_tasks() if not t.completed] + return [t for t in self._index.values() if not t.completed] def get_completed_tasks(self) -> List[Task]: """Get all completed tasks""" - return [t for t in self.load_tasks() if t.completed] \ No newline at end of file + return [t for t in self._index.values() if t.completed] + + + def list_tasks(self, page: int = 1, page_size: int = 20) -> List[Task]: + """Return a page of tasks (pagination for ls command)""" + start = (page - 1) * page_size + end = start + page_size + return list(self._index.values())[start:end] + + def iter_tasks(self, start: int = 0, count: int = 20): + """Yield tasks lazily (for large lists without loading all at once)""" + for t in list(self._index.values())[start:start + count]: + yield t