import sqlite3 def execute_query(db_path: str, query: str): conn = sqlite3.connect(db_path) cursor = conn.cursor() try: cursor.execute(query) # cursor.description is None for non-SELECT or empty results if cursor.description is None: conn.close() return [] columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() conn.close() return [dict(zip(columns, row)) for row in rows] except sqlite3.OperationalError as e: conn.close() raise ValueError(f"SQL execution failed: {e}\nQuery: {query}") except Exception as e: conn.close() raise e