Build Self-Service Analytics SaaS with DuckDB: Users Upload Files, Get Instant Reports
Introduction: Turn Analysis Capability into a Sellable Product
Many data analysts face the same dilemma: spending hours creating reports for clients, only to have them request minor changes that require starting over. Your time is consumed by repetitive work, yet your income is tied to project-based billing.
What if you flipped the model?
Build a self-service analytics platform where clients upload their own CSV, Excel, or Parquet files, select analysis dimensions, click generate, and receive results in seconds. You’re not selling “report creation time”—you’re selling “platform access.”
This is exactly what DuckDB excels at. As an embedded OLAP database with zero configuration, columnar storage, and direct support for multiple file formats, it delivers query performance far beyond traditional approaches.
Why DuckDB for Self-Service Analytics Engines?
| Dimension | PostgreSQL/MySQL | SQLite | DuckDB |
|---|---|---|---|
| Deployment | Requires standalone DB service | Embedded, but row-store | Embedded, column-store |
| Startup Speed | Seconds | Milliseconds | Milliseconds |
| Analytical Query Performance | Moderate (OLTP optimized) | Slow (no vectorized execution) | Excellent (vectorized + columnar) |
| File Format Support | Requires ETL import | Own format only | Native CSV/Parquet/JSON/Excel/HTTP |
| Operational Cost | High (backup, connection pooling, monitoring) | Low | Zero |
| Ideal Use Case | Transactional apps | Small local apps | Analytics/self-service queries |
The core advantage in one sentence: DuckDB gives you an analysis engine as lightweight as SQLite, but with analytical performance rivaling cloud data warehouses.
Complete Implementation: Self-Service Analytics SaaS
Step 1: Install Dependencies
pip install duckdb fastapi uvicorn pandas pydantic python-multipart
Step 2: Create Main Service (main.py)
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)
Step 3: Test the API
# 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'}"
Example Response:
{
"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}
]
}
Advanced: Add User Authentication and Quota Management
Production environments need abuse prevention:
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
💰 Monetization Paths
This architecture can directly convert into multiple business models:
1. Pay-per-use Pricing
- Free tier: 10 free queries/month
- Basic: $99/month, 100 queries
- Pro: $299/month, unlimited queries + Excel export
2. Industry-Specific Templates SaaS
Pre-build analysis templates for different industries:
- E-commerce: Auto-detects orders, products, users tables, generates sales dashboard
- Finance: Auto-detects vouchers, accounts, balances, generates financial reports
- HR: Auto-detects employees, departments, attendance, generates workforce analytics
Each industry template can be sold separately at $500-2000/year.
3. On-Premise Deployment
Help medium/large enterprises deploy to internal networks, charging $5000-20000 per project plus annual maintenance fees.
4. Data Consulting + Platform Lock-in
Offer free data analysis first, then recommend clients integrate their data into your self-service platform for monthly recurring revenue.
⚡ Performance Optimization Tips
- File Caching: Compute hash for identical files to avoid re-parsing
- Connection Pooling: Use persistent
duckdb.connect()connections instead of creating new ones each time - Streaming Processing: For large files, use
read_csv_auto(..., hive_partitioning=true)for batch reading - Result Caching: Cache popular query results for 5-30 minutes
- Async Queues: Place complex queries in Celery queues, return async task IDs
The full version of this article has been published on duckdblab.org, including more detailed steps and reusable production environment templates. Want to systematically learn how to apply DuckDB in enterprise scenarios? duckdblab.org has a complete tutorial series.