Coverage for merco/todo/manager.py: 95%

1"""TodoManager — SQLite 任务管理""" 

2from __future__ import annotations 

3 

4import os 

5import sqlite3 

6import uuid 

7from datetime import datetime 

8from typing import Optional 

9 

10from .models import TodoItem 

11 

12 

13class TodoManager: 

14 """Todo 任务管理器""" 

15 

16 def __init__(self, db_path: str): 

17 self.db_path = os.path.expanduser(db_path) 

18 os.makedirs(os.path.dirname(self.db_path), exist_ok=True) 

19 self._init_db() 

20 

21 def _init_db(self): 

22 with sqlite3.connect(self.db_path) as conn: 

23 conn.execute(""" 

24 CREATE TABLE IF NOT EXISTS todos ( 

25 id TEXT PRIMARY KEY, 

26 title TEXT NOT NULL, 

27 description TEXT DEFAULT '', 

28 status TEXT DEFAULT 'pending', 

29 priority INTEGER DEFAULT 1, 

30 parent_id TEXT, 

31 assigned_to TEXT, 

32 created_at TEXT NOT NULL, 

33 updated_at TEXT NOT NULL, 

34 result TEXT DEFAULT '' 

35 ) 

36 """) 

37 

38 def create(self, title: str, description: str = "", priority: int = 1, parent_id: str = None) -> TodoItem: 

39 """创建任务""" 

40 now = datetime.now().isoformat() 

41 item = TodoItem( 

42 id=str(uuid.uuid4()), 

43 title=title, 

44 description=description, 

45 priority=priority, 

46 parent_id=parent_id, 

47 created_at=now, 

48 updated_at=now, 

49 ) 

50 with sqlite3.connect(self.db_path) as conn: 

51 conn.execute( 

52 "INSERT INTO todos VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", 

53 (item.id, item.title, item.description, item.status, item.priority, 

54 item.parent_id, item.assigned_to, item.created_at, item.updated_at, item.result), 

55 ) 

56 return item 

57 

58 def get(self, id: str) -> Optional[TodoItem]: 

59 """获取任务""" 

60 with sqlite3.connect(self.db_path) as conn: 

61 conn.row_factory = sqlite3.Row 

62 row = conn.execute("SELECT * FROM todos WHERE id = ?", (id,)).fetchone() 

63 if not row: 

64 return None 

65 return self._row_to_item(row) 

66 

67 def update(self, id: str, **kwargs) -> Optional[TodoItem]: 

68 """更新任务""" 

69 item = self.get(id) 

70 if not item: 

71 return None 

72 kwargs["updated_at"] = datetime.now().isoformat() 

73 set_clause = ", ".join(f"{k} = ?" for k in kwargs) 

74 values = list(kwargs.values()) + [id] 

75 with sqlite3.connect(self.db_path) as conn: 

76 conn.execute(f"UPDATE todos SET {set_clause} WHERE id = ?", values) 

77 return self.get(id) 

78 

79 def list(self, status: str = None, parent_id: str = None) -> list[TodoItem]: 

80 """列出任务""" 

81 query = "SELECT * FROM todos WHERE 1=1" 

82 params = [] 

83 if status: 

84 query += " AND status = ?" 

85 params.append(status) 

86 if parent_id: 

87 query += " AND parent_id = ?" 

88 params.append(parent_id) 

89 query += " ORDER BY priority DESC, created_at ASC" 

90 with sqlite3.connect(self.db_path) as conn: 

91 conn.row_factory = sqlite3.Row 

92 rows = conn.execute(query, params).fetchall() 

93 return [self._row_to_item(row) for row in rows] 

94 

95 def delete(self, id: str) -> None: 

96 """删除任务""" 

97 with sqlite3.connect(self.db_path) as conn: 

98 conn.execute("DELETE FROM todos WHERE id = ?", (id,)) 

99 

100 @staticmethod 

101 def _row_to_item(row: sqlite3.Row) -> TodoItem: 

102 return TodoItem( 

103 id=row["id"], 

104 title=row["title"], 

105 description=row["description"], 

106 status=row["status"], 

107 priority=row["priority"], 

108 parent_id=row["parent_id"], 

109 assigned_to=row["assigned_to"], 

110 created_at=row["created_at"], 

111 updated_at=row["updated_at"], 

112 result=row["result"], 

113 )