Build a Personal Data Product Backend with DuckDB in 30 Lines of Code

import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def generate_daily_sales(days=30):
 """Generate 30 days of simulated sales data"""
 np.random.seed(42)
 stores = [f"store_{i:03d}" for i in range(1, 11)]
 products = [f"prod_{i:03d}" for i in range(1, 51)]
 
 records = []
 for day_offset in range(days):
 date = datetime.now() - timedelta(days=day_offset)
 n_records = np.random.randint(15000, 25000)
 
 df = pd.DataFrame({
 'date': np.random.choice(
 pd.date_range(date, periods=1), n_records
 ),
 'store_id': np.random.choice(stores, n_records),
 'product_id': np.random.choice(products, n_records),
 'quantity': np.random.randint(1, 20, n_records),
 'unit_price': np.round(np.random.uniform(10, 500, n_records), 2),
 'region': np.random.choice(['CN', 'US', 'EU'], n_records, p=[0.6, 0.3, 0.1])
 })
 df['revenue'] = df['quantity'] * df['unit_price']
 records.append(df)
 
 return pd.concat(records, ignore_index=True)

# Generate 1.5M rows of simulated data
df = generate_daily_sales(30)
print(f"Generated {len(df)} rows")
print(df.head())

import duckdb

class SalesAnalyticsEngine:
 def __init__(self, parquet_path="s3://bucket/sales/"):
 self.path = parquet_path
 self.con = duckdb.connect("memory:")
 # Scan Parquet directly, zero data copy
 self.con.execute(f"""
 CREATE TABLE sales AS 
 SELECT * FROM read_parquet('{parquet_path}*.parquet')
 """)
 
 def daily_revenue(self, store_id: str, days: int = 30) -> list:
 """Daily revenue for a store over N days"""
 result = self.con.execute(f"""
 SELECT date::DATE as dt,
 SUM(revenue) as revenue,
 SUM(quantity) as orders
 FROM sales 
 WHERE store_id = '{store_id}'
 AND date >= CURRENT_DATE - INTERVAL '{days} DAYS'
 GROUP BY dt
 ORDER BY dt DESC
 LIMIT {days}
 """).fetchall()
 return [{"date": r[0].isoformat(), "revenue": r[1], "orders": r[2]} for r in result]
 
 def top_products(self, store_id: str, n: int = 10) -> list:
 """Top N best-selling products"""
 result = self.con.execute(f"""
 SELECT product_id,
 SUM(quantity) as total_qty,
 SUM(revenue) as total_rev,
 COUNT(DISTINCT date) as active_days
 FROM sales 
 WHERE store_id = '{store_id}'
 AND date >= CURRENT_DATE - INTERVAL '30 DAYS'
 GROUP BY product_id
 ORDER BY total_rev DESC
 LIMIT {n}
 """).fetchall()
 return [{"product_id": r[0], "qty": r[1], "revenue": r[2], "days": r[3]} for r in result]
 
 def regional_breakdown(self, store_id: str) -> dict:
 """Regional sales breakdown"""
 result = self.con.execute(f"""
 SELECT region,
 COUNT(*) as txn_count,
 SUM(revenue) as revenue,
 AVG(unit_price) as avg_price
 FROM sales 
 WHERE store_id = '{store_id}'
 GROUP BY region
 """).fetchall()
 return {r[0]: {"txns": r[1], "revenue": r[2], "avg_price": r[3]} for r in result}

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI(title="E-commerce Sales Analysis API")
engine = SalesAnalyticsEngine()

@app.middleware("http")
async def auth_middleware(request, call_next):
 """Simple API Key authentication"""
 api_key = request.headers.get("X-API-Key")
 if api_key != "your-premium-key":
 raise HTTPException(status_code=401)
 return await call_next(request)

@app.get("/api/v1/{store_id}/daily")
async def get_daily(store_id: str, days: int = 30):
 return {"store_id": store_id, "data": engine.daily_revenue(store_id, days)}

@app.get("/api/v1/{store_id}/top-products")
async def get_top_products(store_id: str, n: int = 10):
 return {"store_id": store_id, "data": engine.top_products(store_id, n)}

@app.get("/api/v1/{store_id}/regional")
async def get_regional(store_id: str):
 return {"store_id": store_id, "data": engine.regional_breakdown(store_id)}

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

import json, redis

r = redis.Redis(host='localhost', port=6379, db=0)

def get_with_cache(store_id: str, query_type: str, ttl: int = 300):
 key = f"sales:{store_id}:{query_type}"
 cached = r.get(key)
 if cached:
 return json.loads(cached)
 
 if query_type == "daily":
 data = engine.daily_revenue(store_id)
 elif query_type == "top_products":
 data = engine.top_products(store_id)
 else:
 data = engine.regional_breakdown(store_id)
 
 r.setex(key, ttl, json.dumps(data))
 return data