Create app.py

app.py

@@ -0,0 +1,318 @@
+import gradio as gr
+from groq import Groq
+from pydantic import BaseModel
+import json
+import sqlite3
+import pandas as pd
+from typing import List, Optional
+import re
+
+# Pydantic Models
+class ValidationStatus(BaseModel):
+    is_valid: bool
+    syntax_errors: list[str]
+
+class SQLQueryGeneration(BaseModel):
+    query: str
+    query_type: str
+    tables_used: list[str]
+    estimated_complexity: str
+    execution_notes: list[str]
+    validation_status: ValidationStatus
+
+class TableSchema(BaseModel):
+    table_name: str
+    columns: list[dict]
+    sample_data: list[dict]
+
+def generate_sample_data(user_query: str, groq_api_key: str) -> dict:
+    """Generate sample table schema and data based on user query"""
+    try:
+        client = Groq(api_key=groq_api_key)
+        
+        # Request to generate table schema and sample data
+        schema_prompt = f"""Based on this query: "{user_query}"
+        
+Generate a realistic database schema with sample data. Return ONLY valid JSON with this structure:
+{{
+  "tables": [
+    {{
+      "table_name": "table_name",
+      "columns": [
+        {{"name": "column_name", "type": "INTEGER|TEXT|REAL|DATE"}},
+        ...
+      ],
+      "sample_data": [
+        {{"column_name": value, ...}},
+        ...at least 10-15 rows
+      ]
+    }}
+  ]
+}}
+
+Make the data realistic and relevant to the query. Include enough variety to make the query results meaningful."""
+
+        response = client.chat.completions.create(
+            model="moonshotai/kimi-k2-instruct-0905",
+            messages=[
+                {"role": "system", "content": "You are a database expert. Generate realistic table schemas and sample data. Return ONLY valid JSON, no markdown formatting."},
+                {"role": "user", "content": schema_prompt}
+            ],
+            temperature=0.7
+        )
+        
+        # Parse response
+        content = response.choices[0].message.content.strip()
+        # Remove markdown code blocks if present
+        content = re.sub(r'```json\s*', '', content)
+        content = re.sub(r'```\s*$', '', content)
+        
+        schema_data = json.loads(content)
+        return schema_data
+    except Exception as e:
+        raise Exception(f"Error generating sample data: {str(e)}")
+
+def create_tables_in_db(schema_data: dict) -> sqlite3.Connection:
+    """Create SQLite tables and populate with sample data"""
+    conn = sqlite3.connect(':memory:')
+    cursor = conn.cursor()
+    
+    for table in schema_data['tables']:
+        table_name = table['table_name']
+        columns = table['columns']
+        
+        # Create table
+        column_defs = []
+        for col in columns:
+            col_type = col['type'].upper()
+            column_defs.append(f"{col['name']} {col_type}")
+        
+        create_table_sql = f"CREATE TABLE {table_name} ({', '.join(column_defs)})"
+        cursor.execute(create_table_sql)
+        
+        # Insert sample data
+        sample_data = table['sample_data']
+        if sample_data:
+            col_names = [col['name'] for col in columns]
+            placeholders = ', '.join(['?' for _ in col_names])
+            insert_sql = f"INSERT INTO {table_name} ({', '.join(col_names)}) VALUES ({placeholders})"
+            
+            for row in sample_data:
+                values = [row.get(col) for col in col_names]
+                cursor.execute(insert_sql, values)
+    
+    conn.commit()
+    return conn
+
+def generate_sql_query(user_query: str, groq_api_key: str, schema_info: str) -> SQLQueryGeneration:
+    """Generate SQL query using Groq API with schema context"""
+    try:
+        client = Groq(api_key=groq_api_key)
+        
+        enhanced_query = f"""Database Schema:
+{schema_info}
+
+User Request: {user_query}
+
+Generate a SQL query that works with the above schema."""
+
+        response = client.chat.completions.create(
+            model="moonshotai/kimi-k2-instruct-0905",
+            messages=[
+                {
+                    "role": "system",
+                    "content": "You are a SQL expert. Generate structured SQL queries from natural language descriptions with proper syntax validation and metadata. Use standard SQL syntax compatible with SQLite.",
+                },
+                {"role": "user", "content": enhanced_query},
+            ],
+            response_format={
+                "type": "json_schema",
+                "json_schema": {
+                    "name": "sql_query_generation",
+                    "schema": SQLQueryGeneration.model_json_schema()
+                }
+            }
+        )
+        
+        sql_query_generation = SQLQueryGeneration.model_validate(
+            json.loads(response.choices[0].message.content)
+        )
+        return sql_query_generation
+    except Exception as e:
+        raise Exception(f"Error generating SQL query: {str(e)}")
+
+def execute_sql_query(conn: sqlite3.Connection, query: str) -> pd.DataFrame:
+    """Execute SQL query and return results as DataFrame"""
+    try:
+        df = pd.read_sql_query(query, conn)
+        return df
+    except Exception as e:
+        raise Exception(f"Error executing SQL query: {str(e)}")
+
+def format_schema_info(schema_data: dict) -> str:
+    """Format schema information for display"""
+    info = []
+    for table in schema_data['tables']:
+        info.append(f"\nTable: {table['table_name']}")
+        info.append("Columns:")
+        for col in table['columns']:
+            info.append(f"  - {col['name']} ({col['type']})")
+        info.append(f"Sample rows: {len(table['sample_data'])}")
+    return '\n'.join(info)
+
+def process_query(user_query: str, groq_api_key: str):
+    """Main processing function"""
+    if not groq_api_key or not groq_api_key.strip():
+        return "❌ Please enter your Groq API key", "", "", "", ""
+    
+    if not user_query or not user_query.strip():
+        return "❌ Please enter a query", "", "", "", ""
+    
+    try:
+        output_log = []
+        
+        # Step 1: Generate sample data
+        output_log.append("### Step 1: Generating Sample Database Schema and Data")
+        output_log.append(f"Query: {user_query}\n")
+        
+        schema_data = generate_sample_data(user_query, groq_api_key)
+        schema_info = format_schema_info(schema_data)
+        
+        output_log.append("✅ Generated database schema:")
+        output_log.append(schema_info)
+        output_log.append("")
+        
+        # Step 2: Create tables
+        output_log.append("### Step 2: Creating In-Memory SQLite Database")
+        conn = create_tables_in_db(schema_data)
+        output_log.append("✅ Tables created and populated with sample data\n")
+        
+        # Display sample data
+        sample_tables_html = []
+        for table in schema_data['tables']:
+            df_sample = pd.DataFrame(table['sample_data'][:5])  # Show first 5 rows
+            sample_tables_html.append(f"<h4>Sample Data from '{table['table_name']}' (first 5 rows):</h4>")
+            sample_tables_html.append(df_sample.to_html(index=False, border=1))
+        
+        # Step 3: Generate SQL query
+        output_log.append("### Step 3: Generating SQL Query")
+        sql_generation = generate_sql_query(user_query, groq_api_key, schema_info)
+        
+        # Format the SQL generation output
+        sql_output = {
+            "query": sql_generation.query,
+            "query_type": sql_generation.query_type,
+            "tables_used": sql_generation.tables_used,
+            "estimated_complexity": sql_generation.estimated_complexity,
+            "execution_notes": sql_generation.execution_notes,
+            "validation_status": {
+                "is_valid": sql_generation.validation_status.is_valid,
+                "syntax_errors": sql_generation.validation_status.syntax_errors
+            }
+        }
+        
+        sql_output_formatted = json.dumps(sql_output, indent=2)
+        output_log.append("✅ SQL Query Generated:\n")
+        
+        # Step 4: Execute query
+        output_log.append("\n### Step 4: Executing SQL Query")
+        output_log.append(f"Executing: {sql_generation.query}\n")
+        
+        result_df = execute_sql_query(conn, sql_generation.query)
+        
+        if len(result_df) == 0:
+            output_log.append("ℹ️ Query executed successfully but returned 0 rows")
+            result_html = "<p>No results found</p>"
+        else:
+            output_log.append(f"✅ Query executed successfully! Returned {len(result_df)} row(s)\n")
+            result_html = result_df.to_html(index=False, border=1)
+        
+        conn.close()
+        
+        # Combine all outputs
+        process_log = '\n'.join(output_log)
+        sample_data_html = '\n'.join(sample_tables_html)
+        
+        return process_log, sql_output_formatted, sample_data_html, result_html, ""
+        
+    except Exception as e:
+        error_msg = f"❌ Error: {str(e)}"
+        return error_msg, "", "", "", ""
+
+# Gradio Interface
+with gr.Blocks(title="SQL Query Generator & Executor", theme=gr.themes.Soft()) as app:
+    gr.Markdown("""
+    # 🔍 SQL Query Generator & Executor
+    
+    This app uses Groq's Kimi-K2 model to:
+    1. Generate realistic sample database tables based on your query
+    2. Generate a structured SQL query from natural language
+    3. Execute the query and show results
+    
+    ### How to use:
+    1. Enter your Groq API key ([Get one here](https://console.groq.com/keys))
+    2. Enter your query in plain English
+    3. Click "Generate & Execute SQL"
+    """)
+    
+    with gr.Row():
+        with gr.Column(scale=2):
+            api_key_input = gr.Textbox(
+                label="Groq API Key",
+                placeholder="Enter your Groq API key here...",
+                type="password"
+            )
+            
+            query_input = gr.Textbox(
+                label="Natural Language Query",
+                placeholder="Example: Find all customers who made orders over $500 in the last 30 days, show their name, email, and total order amount",
+                lines=3
+            )
+            
+            submit_btn = gr.Button("🚀 Generate & Execute SQL", variant="primary", size="lg")
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 📋 Process Log")
+            process_output = gr.Textbox(
+                label="Execution Steps",
+                lines=12,
+                max_lines=20
+            )
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 🗂️ Sample Database Tables")
+            sample_data_output = gr.HTML(label="Sample Data")
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 📝 Generated SQL Query (Structured Output)")
+            sql_output = gr.JSON(label="SQL Query Metadata")
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### ✨ Query Execution Results")
+            result_output = gr.HTML(label="Results")
+    
+    # Examples
+    gr.Examples(
+        examples=[
+            ["Find all customers who made orders over $500 in the last 30 days, show their name, email, and total order amount"],
+            ["List all products that are out of stock along with their supplier information"],
+            ["Show the top 5 employees by total sales in the last quarter"],
+            ["Find all students who scored above 85% in Mathematics and their contact details"],
+            ["Get all active users who haven't logged in for more than 60 days"]
+        ],
+        inputs=query_input,
+        label="Example Queries"
+    )
+    
+    submit_btn.click(
+        fn=process_query,
+        inputs=[query_input, api_key_input],
+        outputs=[process_output, sql_output, sample_data_output, result_output, gr.Textbox(visible=False)]
+    )
+
+if __name__ == "__main__":
+    app.launch()