Build Self-Service Analytics SaaS with DuckDB: Users Upload Files, Get Instant Reports

pip install duckdb fastapi uvicorn pandas pydantic python-multipart

import duckdb
import pandas as pd
from fastapi import FastAPI, UploadFile, File, HTTPException, Query
from pydantic import BaseModel
import tempfile
import os
from typing import Optional, List

app = FastAPI(title="Self-Service Analytics API", version="1.0")

class AnalysisRequest(BaseModel):
 """Analysis request parameters"""
 query_type: str = "auto" # auto, top_n, trend, distribution, custom_sql
 columns: Optional[List[str]] = None # Columns to analyze
 group_by: Optional[str] = None # Group field
 measure: Optional[str] = None # Measure field
 top_k: int = 10 # Top N count
 date_column: Optional[str] = None # Date column
 filter_sql: Optional[str] = None # Custom filter condition

@app.post("/analyze")
async def analyze(
 file: UploadFile = File(...),
 req: AnalysisRequest = AnalysisRequest()
):
 """Core analysis endpoint: upload file + analysis params = instant results"""
 
 # 1. Save uploaded file to temporary directory
 suffix = os.path.splitext(file.filename)[1].lower() if file.filename else '.csv'
 with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
 content = await file.read()
 tmp.write(content)
 tmp_path = tmp.name
 
 try:
 conn = duckdb.connect(database=':memory:')
 
 # 2. Choose read function based on extension
 if suffix == '.csv':
 conn.execute(f"CREATE TABLE data AS SELECT * FROM read_csv_auto('{tmp_path}', header=true)")
 elif suffix in ['.xlsx', '.xls']:
 conn.execute(f"CREATE TABLE data AS SELECT * FROM read_xlsx('{tmp_path}')")
 elif suffix == '.parquet':
 conn.execute(f"CREATE TABLE data AS SELECT * FROM read_parquet('{tmp_path}')")
 elif suffix == '.json':
 conn.execute(f"CREATE TABLE data AS SELECT * FROM read_json_auto('{tmp_path}')")
 else:
 raise HTTPException(status_code=400, detail=f"Unsupported file format: {suffix}")
 
 # 3. Get column info for frontend display
 columns = conn.execute("DESCRIBE data").fetchall()
 column_names = [c[0] for c in columns]
 
 # 4. Execute analysis based on query type
 result = execute_analysis(conn, req, column_names)
 
 conn.close()
 
 return {
 "status": "success",
 "columns": column_names,
 "total_rows": conn.execute("SELECT COUNT(*) FROM data").fetchone()[0],
 "result": result.to_dict('records') if hasattr(result, 'to_dict') else result,
 "preview_sql": result._sql if hasattr(result, '_sql') else ""
 }
 
 except Exception as e:
 raise HTTPException(status_code=500, detail=str(e))
 
 finally:
 if os.path.exists(tmp_path):
 os.unlink(tmp_path)

def execute_analysis(conn: duckdb.DuckDBPyConnection, req: AnalysisRequest, columns: List[str]):
 """Execute different types of analysis queries"""
 
 if req.query_type == 'custom_sql' and req.filter_sql:
 sql = f"SELECT * FROM data WHERE {req.filter_sql} LIMIT 1000"
 return conn.execute(sql).fetchdf()
 
 elif req.query_type == 'top_n':
 group_col = req.group_by or columns[0]
 measure_col = req.measure or "COUNT(*)"
 sql = f"""
 SELECT 
 {group_col},
 COUNT(*) as record_count,
 AVG(CAST({req.measure} AS DOUBLE)) as avg_value
 FROM data 
 GROUP BY {group_col}
 ORDER BY record_count DESC
 LIMIT {req.top_k}
 """
 return conn.execute(sql).fetchdf()
 
 elif req.query_type == 'trend':
 date_col = req.date_column or find_date_column(columns)
 measure_col = req.measure or "SUM(amount)"
 sql = f"""
 SELECT 
 DATE_TRUNC('month', CAST({date_col} AS DATE)) as month,
 COUNT(*) as order_count,
 SUM(CAST({measure_col} AS DOUBLE)) as total_amount
 FROM data
 GROUP BY month
 ORDER BY month
 """
 return conn.execute(sql).fetchdf()
 
 elif req.query_type == 'distribution':
 col = req.columns[0] if req.columns else columns[0]
 sql = f"""
 SELECT 
 {col},
 COUNT(*) as count,
 ROUND(COUNT(*) * 100.0 / (SELECT COUNT(*) FROM data), 2) as percentage
 FROM data
 GROUP BY {col}
 ORDER BY count DESC
 LIMIT {req.top_k}
 """
 return conn.execute(sql).fetchdf()
 
 else: # auto: detect and provide summary
 return get_auto_summary(conn, columns)

def find_date_column(columns: List[str]):
 """Intelligently identify date columns"""
 date_keywords = ['date', 'time', 'created', 'updated', 'order_date']
 for col in columns:
 if any(kw in col.lower() for kw in date_keywords):
 return col
 return columns[0]

def get_auto_summary(conn: duckdb.DuckDBPyConnection, columns: List[str]):
 """Automatically generate data summary"""
 summary_queries = [
 ("Total Rows", "SELECT COUNT(*) FROM data"),
 ("Numeric Column Stats", "SELECT " + ", ".join([f"ROUND(AVG(CAST({c} AS DOUBLE)), 2)" for c in columns if c.isdigit()][:5]) + " FROM data"),
 ]
 
 results = {}
 for name, sql in summary_queries:
 try:
 results[name] = conn.execute(sql).fetchdf().to_dict('records')
 except:
 results[name] = []
 
 return results

if __name__ == "__main__":
 import uvicorn
 uvicorn.run(app, host="0.0.0.0", port=8000)

# Start service
python main.py

# Test with curl
curl -X POST "http://localhost:8000/analyze" \
 -F "file=@sales_data.csv" \
 -F "req={'query_type':'top_n','group_by':'category','top_k':5}"

# Trend analysis
curl -X POST "http://localhost:8000/analyze" \
 -F "file=@sales_data.csv" \
 -F "req={'query_type':'trend','date_column':'order_date','measure':'amount'}"

{
 "status": "success",
 "columns": ["order_id", "category", "amount", "order_date"],
 "total_rows": 15000,
 "result": [
 {"category": "Electronics", "record_count": 3200, "avg_value": 256.8},
 {"category": "Clothing", "record_count": 2800, "avg_value": 189.5}
 ]
}

from fastapi import Depends, Header
import jwt

SECRET_KEY="your-secret-key"

def verify_api_key(x_api_key: str = Header(...)):
 """Simple API Key verification"""
 # In production, validate against database
 allowed_keys = {"demo-key-123", "pro-key-456"}
 if x_api_key not in allowed_keys:
 raise HTTPException(status_code=401, detail="Invalid API Key")
 return x_api_key

@app.post("/analyze")
async def analyze_protected(
 file: UploadFile = File(...),
 req: AnalysisRequest = AnalysisRequest(),
 api_key: str = Depends(verify_api_key)
):
 # ... existing logic ...
 pass