Ask your data questions in plain English

from clickruntime import on_event, credentials, http, ai, notify

PROJECT, DATASET = "my-project", "sales"
TABLES = ["orders", "customers"]

@on_event
def handle(event):
    question = event.json["question"]
    h = {"Authorization": f"Bearer {credentials.connection('google', 'work')}"}
    base = f"https://bigquery.googleapis.com/bigquery/v2/projects/{PROJECT}"
    schemas = []
    for t in TABLES:
        fields = http.get(f"{base}/datasets/{DATASET}/tables/{t}", headers=h).json()["schema"]["fields"]
        schemas.append(f"{DATASET}.{t}(" + ", ".join(f"{f['name']} {f['type']}" for f in fields) + ")")
    sql = ai.generate(f"Tables:\n{chr(10).join(schemas)}\n\nQuestion: {question}",
                      system="Write one BigQuery standard SQL SELECT that answers the question. Reply with SQL only, LIMIT 50.",
                      max_output_tokens=300).text.strip().strip("`").removeprefix("sql").strip()
    if not sql.lower().startswith(("select", "with")):
        raise ValueError("Only SELECT queries are run")
    res = http.post(f"{base}/queries", headers=h, json={"query": sql, "useLegacySql": False, "maxResults": 50}).json()
    cols = [f["name"] for f in res["schema"]["fields"]]
    rows = [" | ".join(str(c["v"]) for c in r["f"]) for r in res.get("rows", [])]
    notify.email_me(f"Answer: {question[:80]}", f"{' | '.join(cols)}\n" + "\n".join(rows) + f"\n\nSQL:\n{sql}")
    return {"rows": len(rows)}