todo-mcp-server/database.py

"""
SQLite database manager for TODO MCP server.
"""

import sqlite3
from datetime import datetime
from typing import List, Dict, Any, Optional
from pathlib import Path
from date_utils import parse_relative_date, get_date_for_query


class TodoDatabase:
    """Manages SQLite database for TODO items"""

    def __init__(self, db_path: str = "todos.db"):
        """
        Initialize database connection.

        Args:
            db_path: Path to SQLite database file
        """
        self.db_path = db_path
        self.conn = None
        self._init_database()

    def _init_database(self):
        """Create database and tables if they don't exist"""
        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row  # Return rows as dictionaries

        cursor = self.conn.cursor()
        cursor.execute(
            """
            CREATE TABLE IF NOT EXISTS todos (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                description TEXT,
                created_date TEXT NOT NULL,
                due_date TEXT,
                status TEXT NOT NULL DEFAULT 'pending',
                completed_date TEXT
            )
        """
        )
        self.conn.commit()

    def create_todo(
        self,
        title: str,
        description: str = "",
        relative_days: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        Create a new TODO item.

        Args:
            title: TODO title
            description: TODO description
            relative_days: Relative due date (0=today, +1=tomorrow, -1=yesterday, etc.)

        Returns:
            Created TODO item as dictionary
        """
        created_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # Convert relative days to absolute date
        due_date = get_date_for_query(relative_days)

        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO todos (title, description, created_date, due_date, status)
            VALUES (?, ?, ?, ?, 'pending')
        """,
            (title, description, created_date, due_date),
        )
        self.conn.commit()

        todo_id = cursor.lastrowid
        return self.get_todo(todo_id)

    def get_todo(self, todo_id: int) -> Optional[Dict[str, Any]]:
        """
        Get a TODO by ID.

        Args:
            todo_id: TODO ID

        Returns:
            TODO item as dictionary or None if not found
        """
        cursor = self.conn.cursor()
        cursor.execute("SELECT * FROM todos WHERE id = ?", (todo_id,))
        row = cursor.fetchone()

        if row:
            return dict(row)
        return None

    def list_todos(
        self,
        relative_days: Optional[int] = None,
        status: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """
        List TODO items with optional filtering.

        Args:
            relative_days: Filter by relative due date (0=today, +1=tomorrow, etc.)
            status: Filter by status ('pending', 'completed')

        Returns:
            List of TODO items
        """
        query = "SELECT * FROM todos WHERE 1=1"
        params = []

        if relative_days is not None:
            # Convert relative days to absolute date for filtering
            date = parse_relative_date(relative_days)
            query += " AND due_date LIKE ?"
            params.append(f"{date}%")

        if status:
            query += " AND status = ?"
            params.append(status)

        query += " ORDER BY due_date ASC, created_date ASC"

        cursor = self.conn.cursor()
        cursor.execute(query, params)
        rows = cursor.fetchall()

        return [dict(row) for row in rows]

    def update_todo(
        self,
        todo_id: int,
        title: Optional[str] = None,
        description: Optional[str] = None,
        relative_days: Optional[int] = None,
    ) -> Optional[Dict[str, Any]]:
        """
        Update a TODO item.

        Args:
            todo_id: TODO ID
            title: New title (optional)
            description: New description (optional)
            relative_days: New relative due date (optional, 0=today, +1=tomorrow, etc.)

        Returns:
            Updated TODO item or None if not found
        """
        # Build dynamic update query
        updates = []
        params = []

        if title is not None:
            updates.append("title = ?")
            params.append(title)

        if description is not None:
            updates.append("description = ?")
            params.append(description)

        if relative_days is not None:
            # Convert relative days to absolute date
            due_date = parse_relative_date(relative_days)
            updates.append("due_date = ?")
            params.append(due_date)

        if not updates:
            return self.get_todo(todo_id)

        params.append(todo_id)
        query = f"UPDATE todos SET {', '.join(updates)} WHERE id = ?"

        cursor = self.conn.cursor()
        cursor.execute(query, params)
        self.conn.commit()

        if cursor.rowcount > 0:
            return self.get_todo(todo_id)
        return None

    def mark_complete(self, todo_id: int) -> Optional[Dict[str, Any]]:
        """
        Mark a TODO as completed.

        Args:
            todo_id: TODO ID

        Returns:
            Updated TODO item or None if not found
        """
        completed_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        cursor = self.conn.cursor()
        cursor.execute(
            """
            UPDATE todos
            SET status = 'completed', completed_date = ?
            WHERE id = ?
        """,
            (completed_date, todo_id),
        )
        self.conn.commit()

        if cursor.rowcount > 0:
            return self.get_todo(todo_id)
        return None

    def delete_todo(self, todo_id: int) -> bool:
        """
        Delete a TODO item.

        Args:
            todo_id: TODO ID

        Returns:
            True if deleted, False if not found
        """
        cursor = self.conn.cursor()
        cursor.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
        self.conn.commit()

        return cursor.rowcount > 0

    def close(self):
        """Close database connection"""
        if self.conn:
            self.conn.close()