Scenario Experience


Feishu
Choose your preferred way to join


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.
No coding required. Ask questions in natural language and AI generates SQL queries and visualizations automatically.
Start your free trial now and experience AI-powered data analysis with AskTable.