AI Natural Language Queries: Ask Your Data Anything

pip install anthropic

[anthropic]
api_key = "sk-ant-..."

[database]
url = "sqlite:///sales.db"

import anthropic

client = anthropic.Anthropic(api_key=st.secrets["anthropic"]["api_key"])

SCHEMA = """
Table: sales
Columns:
  - date TEXT (format: YYYY-MM-DD)
  - region TEXT (values: North, South, East, West)
  - product TEXT (values: Widget A, Widget B, Widget C)
  - revenue INTEGER
  - units INTEGER
"""

def question_to_sql(question: str) -> str:
    """Convert a natural language question to a SQL query."""
    response = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""You are a SQL expert. Given this database schema:

{SCHEMA}

Write a SQLite SELECT query to answer this question:
"{question}"

Rules:
- Return ONLY the SQL query, no explanation
- Use SQLite syntax
- Never use INSERT, UPDATE, DELETE, DROP, or CREATE
- Always include a LIMIT 1000 unless the user asks for all data

SQL query:"""
        }]
    )
    return response.content[0].text.strip()

import re

DANGEROUS = ["insert", "update", "delete", "drop", "create", "alter", "truncate", "exec"]

def is_safe_query(sql: str) -> bool:
    """Return True if the query is a safe SELECT statement."""
    sql_lower = sql.lower().strip()
    # Must start with SELECT
    if not sql_lower.startswith("select"):
        return False
    # Must not contain dangerous keywords
    for keyword in DANGEROUS:
        if re.search(r'\b' + keyword + r'\b', sql_lower):
            return False
    return True

import streamlit as st
import pandas as pd
import plotly.express as px
from sqlalchemy import create_engine, text
import anthropic
import re

st.set_page_config(page_title="Ask AI — Dashboard", layout="wide")

engine = create_engine(st.secrets["database"]["url"])
client = anthropic.Anthropic(api_key=st.secrets["anthropic"]["api_key"])

SCHEMA = """
Table: sales — columns: date (TEXT), region (TEXT), product (TEXT), revenue (INTEGER), units (INTEGER)
"""

DANGEROUS = ["insert", "update", "delete", "drop", "create", "alter", "truncate"]

def is_safe(sql):
    s = sql.lower().strip()
    if not s.startswith("select"):
        return False
    return not any(re.search(r'\b' + k + r'\b', s) for k in DANGEROUS)

def ask_claude(question):
    r = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=500,
        messages=[{"role": "user", "content":
            f"Schema: {SCHEMA}\n\nWrite a SQLite SELECT query for: \"{question}\"\n\n"
            "Return ONLY the SQL query. Include LIMIT 500."}]
    )
    return r.content[0].text.strip()

# ---- UI ----
st.title("Ask Your Data Anything")
st.caption("Powered by Claude — type a question in plain English")

examples = [
    "What is total revenue by region?",
    "Which product sold the most units in January?",
    "Show me monthly revenue for the North region",
    "What are the top 10 highest revenue days?"
]

with st.expander("Example questions"):
    for ex in examples:
        if st.button(ex, key=ex):
            st.session_state.question = ex

question = st.text_input(
    "Your question",
    value=st.session_state.get("question", ""),
    placeholder="e.g. What is total revenue by region?"
)

if question:
    with st.spinner("Generating SQL..."):
        try:
            sql = ask_claude(question)
        except Exception as e:
            st.error(f"API error: {e}")
            st.stop()

    with st.expander("Generated SQL", expanded=False):
        st.code(sql, language="sql")

    if not is_safe(sql):
        st.error("Unsafe query detected. Only SELECT queries are allowed.")
        st.stop()

    try:
        df = pd.read_sql(text(sql), engine)
        st.success(f"{len(df):,} rows returned")
        st.dataframe(df, use_container_width=True)

        # Auto-chart: if there's a numeric column, try to visualize
        numeric_cols = df.select_dtypes("number").columns.tolist()
        text_cols = df.select_dtypes("object").columns.tolist()

        if numeric_cols and text_cols:
            fig = px.bar(df, x=text_cols[0], y=numeric_cols[0],
                         title=question)
            st.plotly_chart(fig, use_container_width=True)

    except Exception as e:
        st.error(f"Query error: {e}")
        st.info("Try rephrasing your question.")