
sidebar.wechat

sidebar.feishu
sidebar.chooseYourWayToJoin

sidebar.scanToAddConsultant
These examples show how to embed AskTable in real apps. Each includes runnable code and notes.
A minimal web app where users ask questions in plain language against your database.
Stack:
Install:
pip install fastapi uvicorn requests python-dotenv
main.py:
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import requests
import os
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# AskTable
ASKTABLE_API_KEY = os.getenv("ASKTABLE_API_KEY")
ASKTABLE_BASE_URL = "https://api.asktable.com/api/v1"
DATASOURCE_ID = os.getenv("DATASOURCE_ID")
headers = {
"Authorization": f"Bearer {ASKTABLE_API_KEY}",
"Content-Type": "application/json"
}
class QueryRequest(BaseModel):
question: str
class QueryResponse(BaseModel):
question: str
sql: str
answer: str
data: list
@app.post("/api/query", response_model=QueryResponse)
async def query_data(request: QueryRequest):
"""Run user query"""
try:
# AskTable API
response = requests.post(
f"{ASKTABLE_BASE_URL}/single-turn/q2a",
headers=headers,
json={
"datasource_id": DATASOURCE_ID,
"question": request.question
}
)
response.raise_for_status()
result = response.json()
return QueryResponse(
question=result["question"],
sql=result["sql"],
answer=result["answer"],
data=result["dataframe"]["data"]
)
except requests.exceptions.HTTPError as e:
raise HTTPException(
status_code=e.response.status_code,
detail=e.response.json().get("detail", "API error")
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/datasources")
async def get_datasources():
"""List datasources"""
try:
response = requests.get(
f"{ASKTABLE_BASE_URL}/datasources",
headers=headers
)
response.raise_for_status()
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
.env:
ASKTABLE_API_KEY=your_api_key_here
DATASOURCE_ID=your_datasource_id_here
Run backend:
python main.py
Create React app:
npx create-react-app data-query-app --template typescript
cd data-query-app
npm install axios
src/App.tsx:
import React, { useState } from 'react';
import axios from 'axios';
import './App.css';
interface QueryResult {
question: string;
sql: string;
answer: string;
data: any[][];
}
function App() {
const [question, setQuestion] = useState('');
const [result, setResult] = useState<QueryResult | null>(null);
const [loading, setLoading] = useState(false);
const [error, setError] = useState('');
const handleSubmit = async (e: React.FormEvent) => {
e.preventDefault();
setLoading(true);
setError('');
setResult(null);
try {
const response = await axios.post('http://localhost:8000/api/query', {
question
});
setResult(response.data);
} catch (err: any) {
setError(err.response?.data?.detail || 'Query failed');
} finally {
setLoading(false);
}
};
return (
<div className="App">
<header className="App-header">
<h1>Data query assistant</h1>
<form onSubmit={handleSubmit}>
<input
type="text"
value={question}
onChange={(e) => setQuestion(e.target.value)}
placeholder="Ask a question, e.g. revenue this month"
disabled={loading}
/>
<button type="submit" disabled={loading}>
{loading ? 'Querying...' : 'Query'}
</button>
</form>
{error && <div className="error">{error}</div>}
{result && (
<div className="result">
<h2>Result</h2>
<div className="answer">
<strong>Answer:</strong>
<p>{result.answer}</p>
</div>
<div className="sql">
<strong>Generated SQL:</strong>
<pre>{result.sql}</pre>
</div>
{result.data.length > 0 && (
<div className="data">
<strong>Rows:</strong>
<table>
<tbody>
{result.data.map((row, i) => (
<tr key={i}>
{row.map((cell, j) => (
<td key={j}>{cell}</td>
))}
</tr>
))}
</tbody>
</table>
</div>
)}
</div>
)}
</header>
</div>
);
}
export default App;
Run frontend:
npm start
Build a Slack bot so the team can query data in natural language from channels.
Features:
DATASOURCE_ID at the right datasourceInstall:
pip install slack-bolt requests python-dotenv
slack_bot.py:
import os
import requests
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from dotenv import load_dotenv
load_dotenv()
# Slack
app = App(token=os.environ.get("SLACK_BOT_TOKEN"))
# AskTable
ASKTABLE_API_KEY = os.environ.get("ASKTABLE_API_KEY")
ASKTABLE_BASE_URL = "https://api.asktable.com/api/v1"
DATASOURCE_ID = os.environ.get("DATASOURCE_ID")
headers = {
"Authorization": f"Bearer {ASKTABLE_API_KEY}",
"Content-Type": "application/json"
}
def query_asktable(question: str) -> dict:
"""Call AskTable Q2A"""
try:
response = requests.post(
f"{ASKTABLE_BASE_URL}/single-turn/q2a",
headers=headers,
json={
"datasource_id": DATASOURCE_ID,
"question": question
},
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
@app.event("app_mention")
def handle_mention(event, say):
"""Handle app_mention"""
# Strip bot mention
text = event["text"]
question = text.split(">", 1)[1].strip() if ">" in text else text
say(f"Querying: _{question}_")
result = query_asktable(question)
if "error" in result:
say(f"❌ Query failed: {result['error']}")
return
# Format blocks
blocks = [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Question:*\n{result['question']}"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Answer:*\n{result['answer']}"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*SQL:*\n```{result['sql']}```"
}
}
]
# Append table
if result.get("dataframe") and result["dataframe"].get("data"):
data = result["dataframe"]["data"]
columns = result["dataframe"]["columns"]
# Table text
table_text = " | ".join(columns) + "\n"
table_text += "-" * (len(table_text) - 1) + "\n"
for row in data[:10]: # Max 10 rows
table_text += " | ".join(str(cell) for cell in row) + "\n"
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Data:*\n```{table_text}```"
}
})
say(blocks=blocks)
@app.command("/query")
def handle_query_command(ack, command, say):
"""Slash command"""
ack()
question = command["text"]
if not question:
say("Usage: /query revenue this month")
return
say(f"Querying: _{question}_")
result = query_asktable(question)
if "error" in result:
say(f"❌ Query failed: {result['error']}")
return
say(f"*Answer:*\n{result['answer']}\n\n*SQL:*\n```{result['sql']}```")
if __name__ == "__main__":
handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
handler.start()
.env:
SLACK_BOT_TOKEN=xoxb-your-bot-token
SLACK_APP_TOKEN=xapp-your-app-token
ASKTABLE_API_KEY=your_api_key_here
DATASOURCE_ID=your_datasource_id_here
Run bot:
python slack_bot.py
app_mentions:read, chat:write, commandsRun a daily job that queries metrics and emails an HTML summary.
Features:
Install:
pip install requests schedule jinja2 python-dotenv
report_generator.py:
import os
import requests
import schedule
import time
from datetime import datetime
from jinja2 import Template
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
from dotenv import load_dotenv
load_dotenv()
# AskTable
ASKTABLE_API_KEY = os.getenv("ASKTABLE_API_KEY")
ASKTABLE_BASE_URL = "https://api.asktable.com/api/v1"
DATASOURCE_ID = os.getenv("DATASOURCE_ID")
# SMTP
SMTP_SERVER = os.getenv("SMTP_SERVER")
SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
SMTP_USER = os.getenv("SMTP_USER")
SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")
REPORT_RECIPIENTS = os.getenv("REPORT_RECIPIENTS").split(",")
headers = {
"Authorization": f"Bearer {ASKTABLE_API_KEY}",
"Content-Type": "application/json"
}
def query_asktable(question: str) -> dict:
"""Call AskTable Q2A"""
try:
response = requests.post(
f"{ASKTABLE_BASE_URL}/single-turn/q2a",
headers=headers,
json={
"datasource_id": DATASOURCE_ID,
"question": question
},
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e), "question": question}
def generate_report():
"""Build report"""
print(f"[{datetime.now()}] Starting report...")
# Questions
questions = [
"Yesterday revenue",
"Yesterday orders",
"Yesterday new users",
"Top 10 products yesterday",
"Revenue by region yesterday"
]
# Run queries
results = []
for question in questions:
print(f" Q: {question}")
result = query_asktable(question)
results.append(result)
# HTML template
html_template = """
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<style>
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
}
h1 {
color: #333;
border-bottom: 2px solid #4CAF50;
padding-bottom: 10px;
}
.metric {
background: #f5f5f5;
padding: 15px;
margin: 10px 0;
border-radius: 5px;
}
.metric h3 {
margin-top: 0;
color: #4CAF50;
}
.answer {
font-size: 18px;
font-weight: bold;
color: #333;
}
.sql {
background: #f0f0f0;
padding: 10px;
border-radius: 3px;
font-family: monospace;
font-size: 12px;
overflow-x: auto;
}
table {
width: 100%;
border-collapse: collapse;
margin-top: 10px;
}
th, td {
border: 1px solid #ddd;
padding: 8px;
text-align: left;
}
th {
background-color: #4CAF50;
color: white;
}
.error {
color: red;
}
</style>
</head>
<body>
<h1>Daily metrics report</h1>
<p>Generated at: {{ report_time }}</p>
{% for result in results %}
<div class="metric">
<h3>{{ result.question }}</h3>
{% if result.error %}
<p class="error">Query failed: {{ result.error }}</p>
{% else %}
<p class="answer">{{ result.answer }}</p>
{% if result.dataframe and result.dataframe.data %}
<table>
<thead>
<tr>
{% for col in result.dataframe.columns %}
<th>{{ col }}</th>
{% endfor %}
</tr>
</thead>
<tbody>
{% for row in result.dataframe.data[:10] %}
<tr>
{% for cell in row %}
<td>{{ cell }}</td>
{% endfor %}
</tr>
{% endfor %}
</tbody>
</table>
{% endif %}
<details>
<summary>Show SQL</summary>
<div class="sql">{{ result.sql }}</div>
</details>
{% endif %}
</div>
{% endfor %}
</body>
</html>
"""
template = Template(html_template)
html_content = template.render(
report_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
results=results
)
# Send email
send_email(html_content)
print(f"[{datetime.now()}] Report done")
def send_email(html_content: str):
"""Send email"""
try:
msg = MIMEMultipart('alternative')
msg['Subject'] = f"Daily metrics report - {datetime.now().strftime('%Y-%m-%d')}"
msg['From'] = SMTP_USER
msg['To'] = ", ".join(REPORT_RECIPIENTS)
html_part = MIMEText(html_content, 'html')
msg.attach(html_part)
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.starttls()
server.login(SMTP_USER, SMTP_PASSWORD)
server.send_message(msg)
print(f" Email sent to: {', '.join(REPORT_RECIPIENTS)}")
except Exception as e:
print(f" Email failed: {e}")
# Schedule
schedule.every().day.at("09:00").do(generate_report)
if __name__ == "__main__":
print("Report scheduler running")
print(f"Daily report at 09:00 to: {', '.join(REPORT_RECIPIENTS)}")
# generate_report() # uncomment for one-off test
# Loop
while True:
schedule.run_pending()
time.sleep(60)
.env:
ASKTABLE_API_KEY=your_api_key_here
DATASOURCE_ID=your_datasource_id_here
SMTP_SERVER=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your_email@gmail.com
SMTP_PASSWORD=your_app_password
REPORT_RECIPIENTS=recipient1@example.com,recipient2@example.com
Run report job:
python report_generator.py
A simple live dashboard for key metrics.
Stack:
dashboard_api.py:
from fastapi import FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddleware
import requests
import asyncio
import os
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# AskTable
ASKTABLE_API_KEY = os.getenv("ASKTABLE_API_KEY")
ASKTABLE_BASE_URL = "https://api.asktable.com/api/v1"
DATASOURCE_ID = os.getenv("DATASOURCE_ID")
headers = {
"Authorization": f"Bearer {ASKTABLE_API_KEY}",
"Content-Type": "application/json"
}
def query_metric(question: str) -> dict:
"""Query one metric"""
try:
response = requests.post(
f"{ASKTABLE_BASE_URL}/single-turn/q2a",
headers=headers,
json={
"datasource_id": DATASOURCE_ID,
"question": question
},
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
@app.get("/api/metrics")
async def get_metrics():
"""All metrics"""
metrics = [
"Today revenue",
"Today orders",
"Today new users",
"Revenue this month",
"Orders this month"
]
results = {}
for metric in metrics:
result = query_metric(metric)
results[metric] = result
return results
@app.websocket("/ws/metrics")
async def websocket_metrics(websocket: WebSocket):
"""WebSocket metrics"""
await websocket.accept()
try:
while True:
# Fetch metrics
metrics = await get_metrics()
# Push to client
await websocket.send_json(metrics)
# Every 30s
await asyncio.sleep(30)
except Exception as e:
print(f"WS error: {e}")
finally:
await websocket.close()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Dashboard.tsx:
import React, { useEffect, useState } from 'react';
import { Chart as ChartJS, ArcElement, Tooltip, Legend } from 'chart.js';
import { Doughnut } from 'react-chartjs-2';
ChartJS.register(ArcElement, Tooltip, Legend);
interface Metrics {
[key: string]: {
answer: string;
dataframe?: {
data: any[][];
};
};
}
function Dashboard() {
const [metrics, setMetrics] = useState<Metrics>({});
const [lastUpdate, setLastUpdate] = useState<Date>(new Date());
useEffect(() => {
// Open WebSocket
const ws = new WebSocket('ws://localhost:8000/ws/metrics');
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
setMetrics(data);
setLastUpdate(new Date());
};
ws.onerror = (error) => {
console.error('WS error:', error);
};
return () => {
ws.close();
};
}, []);
return (
<div className="dashboard">
<h1>Live dashboard</h1>
<p>Last update: {lastUpdate.toLocaleTimeString()}</p>
<div className="metrics-grid">
{Object.entries(metrics).map(([question, result]) => (
<div key={question} className="metric-card">
<h3>{question}</h3>
{result.error ? (
<p className="error">{result.error}</p>
) : (
<p className="value">{result.answer}</p>
)}
</div>
))}
</div>
</div>
);
}
export default Dashboard;
These patterns cover:
The same API can power any surface where users should ask data in natural language.
sidebar.noProgrammingNeeded
sidebar.startFreeTrial