src/db/sql_validator.py - master - yakobsonsa/bi_agent

src/db/sql_validator.py
68 строк2 KB
Yakobsonsa
fix_new
10 фев 2026, 14:55
10 фев 2026, 14:55a4ad889
"""SQL validator implementation."""

import logging
from typing import Dict, Tuple, Any, List
from src.interfaces.sql_validator import ISQLValidator
from src.interfaces.schema_provider import ISchemaProvider

logger = logging.getLogger(__name__)


class SQLValidator(ISQLValidator):
    """
    Concrete SQL validator using schema provider.
    
    Validates SQL queries against database schema metadata
    without direct coupling to database implementation.
    """
    
    def __init__(self, schema_provider: ISchemaProvider):
        """
        Initialize SQL validator.
        
        Args:
            schema_provider: Schema metadata provider
        """
        self.schema_provider = schema_provider
    
    def validate(self, sql: str) -> Dict[str, Any]:
        """
        Validate SQL query against schema.
        
        Delegates to existing validation logic but through interface.
        """
        from src.db.schema_metadata import validate_sql_columns
        return validate_sql_columns(sql)
    
    def fix_hallucinations(self, sql: str) -> Tuple[str, bool]:
        """
        Auto-fix LLM hallucinations in SQL.
        
        Common fixes:
        - Non-existent column names
        - Wrong table references
        """
        from src.db.schema_metadata import fix_sql_hallucinations
        return fix_sql_hallucinations(sql)
    
    def normalize_sql(self, sql: str) -> str:
        """
        Normalize SQL formatting and convert MySQL/PostgreSQL to SQLite syntax.
        
        - Trim whitespace
        - Fix common typos
        - Standardize keywords
        - Convert date functions (MONTH, YEAR, NOW) to SQLite equivalents
        - Fix table/column references
        """
        if not sql:
            return ""
        
        # First apply SQLite dialect conversion
        from src.agent.sql_normalizer import SQLNormalizer
        normalized = SQLNormalizer.normalize_query_for_sqlite(sql)
        
        # Then apply standard normalizations
        normalized = normalized.strip()
        
        return normalized