DuckDB Multi-tenant Financial Report SaaS: From Single Client to Scalable Revenue

import duckdb
from pathlib import Path
from typing import Optional

class TenantDatabase:
 """Multi-tenant DuckDB database manager"""
 
 def __init__(self, db_root: str = "customers"):
 self.db_root = Path(db_root)
 self.db_root.mkdir(exist_ok=True)
 self.master_conn = duckdb.connect("master.duckdb")
 self._init_master()
 
 def _init_master(self):
 """Initialize metadata table"""
 self.master_conn.execute("""
 CREATE TABLE IF NOT EXISTS tenants (
 id VARCHAR PRIMARY KEY,
 name VARCHAR,
 plan VARCHAR DEFAULT 'basic',
 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 data_files VARCHAR[]
 )
 """)
 
 def create_tenant(self, tenant_id: str, name: str) -> str:
 """Create isolated database for tenant"""
 db_path = self.db_root / f"{tenant_id}.duckdb"
 
 # Create independent database
 con = duckdb.connect(str(db_path))
 con.execute("""
 CREATE TABLE income (
 date DATE,
 category VARCHAR,
 amount DECIMAL(10,2),
 description VARCHAR
 )
 """)
 con.execute("""
 CREATE TABLE expense (
 date DATE,
 category VARCHAR,
 amount DECIMAL(10,2),
 vendor VARCHAR
 )
 """)
 con.close()
 
 # Record metadata
 self.master_conn.execute(
 "INSERT INTO tenants (id, name) VALUES (?, ?)",
 (tenant_id, name)
 )
 
 return str(db_path)
 
 def get_conn(self, tenant_id: str) -> duckdb.DuckDBPyConnection:
 """Get tenant database connection"""
 db_path = self.db_root / f"{tenant_id}.duckdb"
 return duckdb.connect(str(db_path))

from datetime import datetime
import json

class ReportGenerator:
 """Report generation engine"""
 
 def __init__(self, db_manager: TenantDatabase):
 self.db = db_manager
 
 def generate_monthly_report(self, tenant_id: str, year: int, month: int) -> dict:
 """Generate single tenant monthly report"""
 conn = self.db.get_conn(tenant_id)
 
 period = f"{year}-{month:02d}"
 
 # Income summary
 income = conn.execute(f"""
 SELECT 
 category,
 SUM(amount) as total,
 COUNT(*) as transactions
 FROM income
 WHERE strftime(date, '%Y-%m') = '{period}'
 GROUP BY category
 ORDER BY total DESC
 """).fetchall()
 
 # Expense summary
 expense = conn.execute(f"""
 SELECT 
 category,
 SUM(amount) as total,
 COUNT(*) as transactions
 FROM expense
 WHERE strftime(date, '%Y-%m') = '{period}'
 GROUP BY category
 ORDER BY total DESC
 """).fetchall()
 
 # Key metrics
 metrics = conn.execute(f"""
 SELECT 
 SUM(CASE WHEN type='income' THEN amount ELSE 0 END) as total_income,
 SUM(CASE WHEN type='expense' THEN amount ELSE 0 END) as total_expense,
 SUM(CASE WHEN type='income' THEN amount ELSE 0 END) - 
 SUM(CASE WHEN type='expense' THEN amount ELSE 0 END) as net_profit,
 ROUND(
 (SUM(CASE WHEN type='income' THEN amount ELSE 0 END) - 
 SUM(CASE WHEN type='expense' THEN amount ELSE 0 END)) * 100.0 /
 NULLIF(SUM(CASE WHEN type='income' THEN amount ELSE 0 END), 0),
 2
 ) as profit_margin
 FROM (
 SELECT date, 'income' as type, amount FROM income
 UNION ALL
 SELECT date, 'expense' as type, amount FROM expense
 ) t
 WHERE strftime(t.date, '%Y-%m') = '{period}'
 """).fetchone()
 
 conn.close()
 
 return {
 'tenant_id': tenant_id,
 'period': period,
 'generated_at': datetime.now().isoformat(),
 'income': [{'category': r[0], 'total': r[1], 'count': r[2]} for r in income],
 'expense': [{'category': r[0], 'total': r[1], 'count': r[2]} for r in expense],
 'metrics': {
 'total_income': metrics[0] or 0,
 'total_expense': metrics[1] or 0,
 'net_profit': metrics[2] or 0,
 'profit_margin': metrics[3] or 0
 }
 }
 
 def generate_all_tenants(self, year: int, month: int):
 """Batch generate reports for all tenants"""
 tenants = self.db.master_conn.execute(
 "SELECT id FROM tenants"
 ).fetchall()
 
 results = []
 for (tenant_id,) in tenants:
 try:
 report = self.generate_monthly_report(tenant_id, year, month)
 results.append(report)
 except Exception as e:
 print(f"Tenant {tenant_id} failed: {e}")
 results.append({'tenant_id': tenant_id, 'error': str(e)})
 
 # Save summary
 output_path = Path(f"reports/{year}_{month:02d}_batch.json")
 output_path.parent.mkdir(exist_ok=True)
 output_path.write_text(json.dumps(results, indent=2, ensure_ascii=False))
 
 return results

from fastapi import FastAPI, UploadFile, File, HTTPException
import asyncio

app = FastAPI(title="DuckDB Financial SaaS")
db_manager = TenantDatabase()
report_gen = ReportGenerator(db_manager)

@app.post("/api/{tenant_id}/upload")
async def upload_data(tenant_id: str, file: UploadFile = File(...)):
 """Upload CSV/Excel data"""
 conn = db_manager.get_conn(tenant_id)
 
 filename = file.filename.lower()
 if filename.endswith('.csv'):
 content = await file.read()
 df = conn.sql(f"SELECT * FROM read_csv_auto(STRING_SPLIT({content.decode()!r}, E'\\n'))")
 elif filename.endswith(('.xlsx', '.xls')):
 df = conn.sql(f"SELECT * FROM read_xlsx('/tmp/{tenant_id}.xlsx')")
 
 table_name = df.columns[0]
 conn.execute(f"INSERT INTO {table_name} SELECT * FROM df")
 conn.close()
 
 return {"status": "success", "rows": len(df)}

@app.post("/api/{tenant_id}/report")
async def generate_report(tenant_id: str, year: int, month: int):
 """Generate single tenant report"""
 try:
 report = report_gen.generate_monthly_report(tenant_id, year, month)
 return report
 except Exception as e:
 raise HTTPException(status_code=404, detail=str(e))

@app.post("/api/scheduler/generate")
async def batch_generate(year: int, month: int):
 """Batch generate all tenant reports"""
 results = await asyncio.to_thread(
 report_gen.generate_all_tenants, year, month
 )
 return {"total": len(results), "success": len([r for r in results if 'error' not in r])}

# schedule.py
import schedule
import time

def daily_report_job():
 """Batch generate on 1st of each month at 2 AM"""
 now = datetime.now()
 if now.day == 1:
 results = report_gen.generate_all_tenants(now.year, now.month)
 send_notification(results)

schedule.every().day.at("02:00").do(daily_report_job)

while True:
 schedule.run_pending()
 time.sleep(60)

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
RUN mkdir -p customers reports

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

-- Query Parquet data lake with DuckDB
SELECT 
 tenant_id,
 date_trunc('month', date) as month,
 SUM(amount) as total
FROM read_parquet('s3://bucket/reports/*.parquet')
GROUP BY 1, 2
ORDER BY 1, 2;