feat(storage): add indexing, pagination, and lazy loading to JSON TaskStorage

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)

tix/storage/json_storage.py

@@ -5,14 +5,19 @@
 
 
 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"""
         self.storage_path = storage_path or (Path.home() / ".tix" / "tasks.json")
         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]
+        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