import sqlite3 from typing import List, Dict, Any import logging class SQLGenerator: def __init__(self, db_path: str = "shopify.db"): self.db_path = db_path self.setup_logging() def setup_logging(self): logging.basicConfig(level=logging.INFO) self.logger = logging.getLogger(__name__) def execute_query(self, query: str) -> List[Dict[str, Any]]: """ Execute SQL query and return results as a list of dictionaries """ try: with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(query) results = [dict(row) for row in cursor.fetchall()] self.logger.info(f"Successfully executed query: {query[:100]}...") return results except sqlite3.Error as e: self.logger.error(f"Database error: {e}") raise except Exception as e: self.logger.error(f"Error executing query: {e}") raise def validate_query(self, query: str) -> bool: """ Validate SQL query before execution """ # Basic validation - you might want to add more sophisticated validation dangerous_keywords = ["DROP", "DELETE", "TRUNCATE", "UPDATE", "INSERT"] return not any(keyword in query.upper() for keyword in dangerous_keywords)