crew/tools/db_utils.py

import sqlite3
from typing import Any, List, Tuple


def run_query(db_path: str, query: str, params: Tuple = ()) -> List[Tuple[Any, ...]]:
    """
    Runs a safe SQL query against a SQLite database and returns the result.

    Args:
        db_path (str): Full path to the SQLite db file.
        query (str): SQL query string to execute.
        params (tuple): Parameters for use in a parameterized SQL statement.

    Returns:
        List[Tuple]: Query results as rows (tuples). Empty list if no results or error.
    """

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

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

        conn.commit()
        conn.close()

        return rows

    except sqlite3.Error as e:
        print(f"[DB ERROR] {e}")
        return []