agent-todo-tracker/agent_todos/store.py

import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Optional

from .models import Priority, Status, Todo

SCHEMA = """
CREATE TABLE IF NOT EXISTS todos (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    repo TEXT DEFAULT '',
    status TEXT NOT NULL DEFAULT 'open',
    priority TEXT NOT NULL DEFAULT 'medium',
    assignee TEXT DEFAULT '',
    tags TEXT DEFAULT '',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    resolved_at TEXT DEFAULT NULL
);
CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status);
CREATE INDEX IF NOT EXISTS idx_todos_repo ON todos(repo);
"""


class Store:
    def __init__(self, db_path: str = ".agent_todos.db") -> None:
        self.db_path = db_path
        self._ensure()

    def _ensure(self) -> None:
        first = not Path(self.db_path).exists()
        with sqlite3.connect(self.db_path) as con:
            con.executescript(SCHEMA)
            con.commit()
        if first:
            self.seed()

    def seed(self) -> None:
        now = datetime.utcnow().isoformat() + "Z"
        self.create(Todo(
            id=None,
            title="Scaffold agent-todo-tracker CLI",
            repo="stackchain/agent-todo-tracker",
            status=Status.resolved,
            priority=Priority.high,
            tags=["hackathon", "p2"],
            created_at=now,
            updated_at=now,
            resolved_at=now,
        ))
        self.create(Todo(
            id=None,
            title="Add Gitea issue sync for TODOs",
            repo="stackchain/stackchain-hackathon",
            status=Status.open,
            priority=Priority.medium,
            tags=["hackathon", "p1"],
            created_at=now,
            updated_at=now,
        ))

    def create(self, todo: Todo) -> Todo:
        now = datetime.utcnow().isoformat() + "Z"
        todo.created_at = now
        todo.updated_at = now
        with sqlite3.connect(self.db_path) as con:
            cur = con.execute(
                "INSERT INTO todos (title, repo, status, priority, assignee, tags, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    todo.title,
                    todo.repo,
                    todo.status.value,
                    todo.priority.value,
                    todo.assignee,
                    ",".join(todo.tags),
                    todo.created_at,
                    todo.updated_at,
                ),
            )
            todo.id = cur.lastrowid
            con.commit()
        return todo

    def list(self, status: Optional[Status] = None, repo: Optional[str] = None) -> list[Todo]:
        query = "SELECT * FROM todos WHERE 1=1"
        params: list = []
        if status:
            query += " AND status = ?"
            params.append(status.value)
        if repo:
            query += " AND repo = ?"
            params.append(repo)
        query += " ORDER BY created_at DESC"
        with sqlite3.connect(self.db_path) as con:
            con.row_factory = sqlite3.Row
            rows = con.execute(query, params).fetchall()
        return [self._row_to_todo(r) for r in rows]

    def update(self, todo_id: int, **kwargs) -> Optional[Todo]:
        todo = self.get(todo_id)
        if not todo:
            return None
        allowed = {"title", "repo", "status", "priority", "assignee", "tags", "resolved_at"}
        fields = {k: v for k, v in kwargs.items() if k in allowed}
        if not fields:
            return todo
        now = datetime.utcnow().isoformat() + "Z"
        sets = ["updated_at = ?"]
        params = [now]
        for k, v in fields.items():
            if k == "tags" and isinstance(v, list):
                v = ",".join(v)
            if k == "status":
                v = v.value if isinstance(v, Status) else v
            if k == "priority":
                v = v.value if isinstance(v, Priority) else v
            sets.append(f"{k} = ?")
            params.append(v)
        if fields.get("status") == Status.resolved.value:
            sets.append("resolved_at = ?")
            params.append(now)
        params.append(todo_id)
        with sqlite3.connect(self.db_path) as con:
            con.execute(f"UPDATE todos SET {', '.join(sets)} WHERE id = ?", params)
            con.commit()
        return self.get(todo_id)

    def get(self, todo_id: int) -> Optional[Todo]:
        with sqlite3.connect(self.db_path) as con:
            con.row_factory = sqlite3.Row
            row = con.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
        return self._row_to_todo(row) if row else None

    def _row_to_todo(self, row: sqlite3.Row) -> Todo:
        return Todo(
            id=row["id"],
            title=row["title"],
            repo=row["repo"],
            status=Status(row["status"]),
            priority=Priority(row["priority"]),
            assignee=row["assignee"],
            tags=[t for t in row["tags"].split(",") if t],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
            resolved_at=row["resolved_at"],
        )