refactor(services): enhance TodoService with callbacks, UUID IDs, and UTF‑8 file handling

This commit is contained in:
sankalp committed 2026-07-12 09:33:41 +05:45
1 parent 51283125ef
commit afd826d0c0
6 files changed
+440 -44

No files matched your search

+43 -36
View File
@@ -1,30 +1,42 @@
# Standard library imports
import json
import os
import uuid
from datetime import datetime
from pathlib import Path
from fabric.core.service import Service
from fabric.utils import logger
# Fabric imports
from fabric.core.service import Property, Service
# Local imports
import shared.data as data
class TodoService(Service):
"""Service for managing persistent todo list with JSON storage"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
def __init__(self):
super().__init__()
self._todos = []
self._file_path = self._get_todos_file_path()
self._load_todos()
self._callbacks = []
def _notify_callbacks(self, event: str, data=None):
def add_callback(self, callback):
"""Add a callback function to be notified of changes"""
self._callbacks.append(callback)
def remove_callback(self, callback):
"""Remove a callback function"""
if callback in self._callbacks:
self._callbacks.remove(callback)
def _notify_callbacks(self, event_type, data=None):
"""Notify all registered callbacks of changes"""
for callback in self._callbacks:
try:
callback(event, data)
callback(event_type, data)
except Exception as e:
logger.error(f"Error in todo callback: {e}")
print(f"Error in todo callback: {e}")
def _get_todos_file_path(self):
"""Returns the path to the todos JSON file"""
@@ -36,23 +48,23 @@ class TodoService(Service):
"""Load todos from JSON file"""
try:
if self._file_path.exists():
with open(self._file_path, "r") as f:
with open(self._file_path, "r", encoding="utf-8") as f:
self._todos = json.load(f)
else:
self._todos = []
except Exception as e:
logger.error(f"Error loading todos: {e}")
print(f"Error loading todos: {e}")
self._todos = []
def _save_todos(self):
"""Save todos to JSON file"""
try:
with open(self._file_path, "w") as f:
with open(self._file_path, "w", encoding="utf-8") as f:
json.dump(self._todos, f, indent=2, ensure_ascii=False)
except Exception as e:
logger.error(f"Error saving todos: {e}")
print(f"Error saving todos: {e}")
@property
@Property(list, "readable")
def todos(self):
"""Get all todos"""
return self._todos.copy()
@@ -60,10 +72,10 @@ class TodoService(Service):
def add_todo(self, text: str, priority: str = "medium") -> dict:
"""Add a new todo item"""
todo = {
"id": os.urandom(8).hex(),
"id": str(uuid.uuid4()),
"text": text,
"completed": False,
"priority": priority,
"priority": priority, # low, medium, high
"created_at": datetime.now().isoformat(),
"updated_at": datetime.now().isoformat(),
}
@@ -110,11 +122,7 @@ class TodoService(Service):
def set_priority(self, todo_id: str, priority: str) -> bool:
"""Set the priority of a todo item"""
valid_priorities = ["low", "medium", "high"]
if priority not in valid_priorities:
logger.warning(
f"Invalid priority '{priority}'. Must be one of {valid_priorities}"
)
if priority not in ["low", "medium", "high"]:
return False
for todo in self._todos:
@@ -127,6 +135,13 @@ class TodoService(Service):
return True
return False
def get_todo(self, todo_id: str) -> dict | None:
"""Get a specific todo by ID"""
for todo in self._todos:
if todo["id"] == todo_id:
return todo.copy()
return None
def clear_completed(self):
"""Remove all completed todos"""
initial_count = len(self._todos)
@@ -139,28 +154,20 @@ class TodoService(Service):
"""Get todo statistics"""
total = len(self._todos)
completed = sum(1 for todo in self._todos if todo["completed"])
pending = total - completed
return {
"total": total,
"completed": completed,
"active": total - completed,
"pending": pending,
"completion_rate": (completed / total * 100) if total > 0 else 0,
}
def add_callback(self, callback):
"""Add a callback for todo events"""
if callback not in self._callbacks:
self._callbacks.append(callback)
def remove_callback(self, callback):
"""Remove a callback"""
if callback in self._callbacks:
self._callbacks.remove(callback)
_todo_service = None
# Global service instance
todo_service = TodoService()
def get_todo_service() -> TodoService:
global _todo_service
if _todo_service is None:
_todo_service = TodoService()
return _todo_service
"""Get the global todo service instance"""
return todo_service