getDiscography/python/utils/archive/user_sqlite.py

import sqlite3
from typing import Any, Iterable, List, Tuple, Optional, Dict


class SQLiteDB:
    """Thin wrapper around sqlite3 that exposes CRUD helpers."""

    def __init__(self, db_path: str = ":memory:") -> None:
        """
        Open a connection to the SQLite database.

        Parameters
        ----------
        db_path : str
            Path to the database file. Use ':memory:' for an in‑memory DB.
        """
        self.db_path = db_path
        self.conn: sqlite3.Connection = sqlite3.connect(self.db_path)
        # Row factory makes rows behave like dictionaries
        self.conn.row_factory = sqlite3.Row
        self.cursor: sqlite3.Cursor = self.conn.cursor()

    # ------------------------------------------------------------------
    # Table helpers
    # ------------------------------------------------------------------
    def create_users_table(self) -> None:
        """
        Create a simple 'users' table if it does not exist.

        Table schema:
            id   INTEGER PRIMARY KEY AUTOINCREMENT
            name TEXT NOT NULL
            age  INTEGER
        """
        sql = """
        CREATE TABLE IF NOT EXISTS users (
            id   INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            age  INTEGER
        );
        """
        self.cursor.execute(sql)
        self.conn.commit()

    # ------------------------------------------------------------------
    # CRUD
    # ------------------------------------------------------------------
    def insert_user(self, name: str, age: Optional[int] = None) -> int:
        """
        Insert a new user.

        Returns
        -------
        int
            The newly created row id.
        """
        sql = "INSERT INTO users (name, age) VALUES (?, ?)"
        self.cursor.execute(sql, (name, age))
        self.conn.commit()
        return self.cursor.lastrowid

    def get_user(self, user_id: int) -> Optional[Dict[str, Any]]:
        """
        Retrieve a single user by id.

        Returns
        -------
        dict or None
            Row as a dictionary if found, otherwise None.
        """
        sql = "SELECT * FROM users WHERE id = ?"
        self.cursor.execute(sql, (user_id,))
        row = self.cursor.fetchone()
        return dict(row) if row else None

    def get_all_users(self) -> List[Dict[str, Any]]:
        """
        Retrieve all users.

        Returns
        -------
        list of dict
            All rows as dictionaries.
        """
        sql = "SELECT * FROM users"
        self.cursor.execute(sql)
        rows = self.cursor.fetchall()
        return [dict(row) for row in rows]

    def update_user(self, user_id: int, name: Optional[str] = None,
                    age: Optional[int] = None) -> bool:
        """
        Update a user's name and/or age.

        Parameters
        ----------
        user_id : int
            The id of the user to update.
        name : str | None
            New name (or None to leave unchanged).
        age : int | None
            New age (or None to leave unchanged).

        Returns
        -------
        bool
            True if a row was updated, False otherwise.
        """
        # Build dynamic SET clause
        fields: List[str] = []
        values: List[Any] = []

        if name is not None:
            fields.append("name = ?")
            values.append(name)
        if age is not None:
            fields.append("age = ?")
            values.append(age)

        if not fields:
            # Nothing to update
            return False

        values.append(user_id)
        sql = f"UPDATE users SET {', '.join(fields)} WHERE id = ?"
        self.cursor.execute(sql, tuple(values))
        self.conn.commit()
        return self.cursor.rowcount > 0

    def delete_user(self, user_id: int) -> bool:
        """
        Delete a user by id.

        Returns
        -------
        bool
            True if a row was deleted, False otherwise.
        """
        sql = "DELETE FROM users WHERE id = ?"
        self.cursor.execute(sql, (user_id,))
        self.conn.commit()
        return self.cursor.rowcount > 0

    # ------------------------------------------------------------------
    # Utility
    # ------------------------------------------------------------------
    def close(self) -> None:
        """Close the database connection."""
        self.cursor.close()
        self.conn.close()

    # Context manager support
    def __enter__(self) -> "SQLiteDB":
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self.close()