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;