1+ """Todo storage utilities for managing todo items."""
2+
3+ import json
4+ import os
5+ from typing import List , Dict , Any , Optional
6+ from dataclasses import dataclass , asdict
7+ from enum import Enum
8+
9+
10+ class TodoStatus (Enum ):
11+ PENDING = "pending"
12+ IN_PROGRESS = "in_progress"
13+ COMPLETED = "completed"
14+
15+
16+ class TodoPriority (Enum ):
17+ HIGH = "high"
18+ MEDIUM = "medium"
19+ LOW = "low"
20+
21+
22+ @dataclass
23+ class TodoItem :
24+ id : str
25+ content : str
26+ status : TodoStatus
27+ priority : TodoPriority
28+
29+ def to_dict (self ) -> Dict [str , Any ]:
30+ return {
31+ "id" : self .id ,
32+ "content" : self .content ,
33+ "status" : self .status .value ,
34+ "priority" : self .priority .value
35+ }
36+
37+ @classmethod
38+ def from_dict (cls , data : Dict [str , Any ]) -> 'TodoItem' :
39+ return cls (
40+ id = data ["id" ],
41+ content = data ["content" ],
42+ status = TodoStatus (data ["status" ]),
43+ priority = TodoPriority (data ["priority" ])
44+ )
45+
46+
47+ class TodoStorage :
48+ def __init__ (self , storage_dir : str = ".minion_workspace" ):
49+ self .storage_dir = storage_dir
50+ os .makedirs (storage_dir , exist_ok = True )
51+
52+ def _get_file_path (self , agent_id : Optional [str ] = None ) -> str :
53+ filename = f"todos_{ agent_id } .json" if agent_id else "todos_default.json"
54+ return os .path .join (self .storage_dir , filename )
55+
56+ def get_todos (self , agent_id : Optional [str ] = None ) -> List [TodoItem ]:
57+ """Get all todos for a specific agent or default."""
58+ file_path = self ._get_file_path (agent_id )
59+
60+ if not os .path .exists (file_path ):
61+ return []
62+
63+ try :
64+ with open (file_path , 'r' , encoding = 'utf-8' ) as f :
65+ data = json .load (f )
66+ return [TodoItem .from_dict (item ) for item in data ]
67+ except (json .JSONDecodeError , KeyError , ValueError ):
68+ return []
69+
70+ def set_todos (self , todos : List [TodoItem ], agent_id : Optional [str ] = None ) -> None :
71+ """Set todos for a specific agent or default."""
72+ file_path = self ._get_file_path (agent_id )
73+
74+ data = [todo .to_dict () for todo in todos ]
75+
76+ with open (file_path , 'w' , encoding = 'utf-8' ) as f :
77+ json .dump (data , f , indent = 2 , ensure_ascii = False )
78+
79+ def add_todo (self , todo : TodoItem , agent_id : Optional [str ] = None ) -> None :
80+ """Add a new todo."""
81+ todos = self .get_todos (agent_id )
82+ todos .append (todo )
83+ self .set_todos (todos , agent_id )
84+
85+ def update_todo (self , todo_id : str , updates : Dict [str , Any ], agent_id : Optional [str ] = None ) -> bool :
86+ """Update a specific todo. Returns True if found and updated."""
87+ todos = self .get_todos (agent_id )
88+
89+ for todo in todos :
90+ if todo .id == todo_id :
91+ if 'content' in updates :
92+ todo .content = updates ['content' ]
93+ if 'status' in updates :
94+ todo .status = TodoStatus (updates ['status' ])
95+ if 'priority' in updates :
96+ todo .priority = TodoPriority (updates ['priority' ])
97+
98+ self .set_todos (todos , agent_id )
99+ return True
100+
101+ return False
102+
103+ def remove_todo (self , todo_id : str , agent_id : Optional [str ] = None ) -> bool :
104+ """Remove a todo by ID. Returns True if found and removed."""
105+ todos = self .get_todos (agent_id )
106+ original_length = len (todos )
107+
108+ todos = [todo for todo in todos if todo .id != todo_id ]
109+
110+ if len (todos ) < original_length :
111+ self .set_todos (todos , agent_id )
112+ return True
113+
114+ return False
115+
116+ def clear_todos (self , agent_id : Optional [str ] = None ) -> None :
117+ """Clear all todos for a specific agent."""
118+ self .set_todos ([], agent_id )
119+
120+
121+ # Global storage instance
122+ _storage = TodoStorage ()
123+
124+ def get_todos (agent_id : Optional [str ] = None ) -> List [TodoItem ]:
125+ """Get todos from global storage."""
126+ return _storage .get_todos (agent_id )
127+
128+ def set_todos (todos : List [TodoItem ], agent_id : Optional [str ] = None ) -> None :
129+ """Set todos in global storage."""
130+ _storage .set_todos (todos , agent_id )
131+
132+ def add_todo (todo : TodoItem , agent_id : Optional [str ] = None ) -> None :
133+ """Add todo to global storage."""
134+ _storage .add_todo (todo , agent_id )
135+
136+ def update_todo (todo_id : str , updates : Dict [str , Any ], agent_id : Optional [str ] = None ) -> bool :
137+ """Update todo in global storage."""
138+ return _storage .update_todo (todo_id , updates , agent_id )
139+
140+ def remove_todo (todo_id : str , agent_id : Optional [str ] = None ) -> bool :
141+ """Remove todo from global storage."""
142+ return _storage .remove_todo (todo_id , agent_id )
143+
144+ def clear_todos (agent_id : Optional [str ] = None ) -> None :
145+ """Clear all todos in global storage."""
146+ _storage .clear_todos (agent_id )
0 commit comments