Add a task to the queue. Returns task ID.
(self, description: str, dependencies: list = None)
| 183 | # ==================== Task Queue ==================== |
| 184 | |
| 185 | def add_task(self, description: str, dependencies: list = None) -> int: |
| 186 | """Add a task to the queue. Returns task ID.""" |
| 187 | import json |
| 188 | deps = json.dumps(dependencies or []) |
| 189 | with self._lock: |
| 190 | cur = self._conn.execute( |
| 191 | "INSERT INTO task_queue (description, status, created_at, dependencies) VALUES (?, 'pending', ?, ?)", |
| 192 | (description, int(time.time()), deps), |
| 193 | ) |
| 194 | self._conn.commit() |
| 195 | return cur.lastrowid |
| 196 | |
| 197 | def get_task(self, task_id: int) -> Optional[Dict]: |
| 198 | """Get a task by ID.""" |
no test coverage detected