Method get_task

core/state.py:197–202

Get a task by ID.

(self, task_id: int)

            return cur.lastrowid

    def get_task(self, task_id: int) -> Optional[Dict]:
        """Get a task by ID."""
        with self._lock:
            cur = self._conn.execute("SELECT * FROM task_queue WHERE id = ?", (task_id,))
            row = cur.fetchone()
            return dict(row) if row else None

    def get_next_pending(self) -> Optional[Dict]:
        """Get the next pending task whose dependencies are all done (oldest first).