# Prototype code: New connection per request
@app.get("/api/sales")
def get_sales():
 con = duckdb.connect("sales.db")
 df = con.execute("SELECT * FROM sales").df()
 con.close()
 return df.to_dict(orient="records")

# database.py — Production-grade connection manager
import duckdb
import threading
from contextlib import contextmanager

class ConnectionPool:
 """DuckDB connection pool with thread-safe access"""
 
 def __init__(self, db_path, pool_size=10):
 self.db_path = db_path
 self.pool_size = pool_size
 self._lock = threading.Lock()
 self._connections = []
 self._in_use = set()
 
 # Pre-create connection pool
 for _ in range(pool_size):
 conn = duckdb.connect(db_path)
 conn.execute("SET memory_limit='4GB';")
 conn.execute("SET threads=4;")
 conn.execute("SET parquet_persistent_cache='true';")
 self._connections.append(conn)
 
 def get_connection(self):
 """Get a connection from the pool (thread-safe)"""
 with self._lock:
 if not self._connections:
 if len(self._connections) + len(self._in_use) < self.pool_size * 2:
 conn = duckdb.connect(self.db_path)
 conn.execute("SET memory_limit='4GB';")
 conn.execute("SET threads=4;")
 self._connections.append(conn)
 else:
 raise RuntimeError("Connection pool full")
 
 conn = self._connections.pop()
 self._in_use.add(id(conn))
 return conn
 
 def release_connection(self, conn):
 """Return connection to the pool"""
 with self._lock:
 self._in_use.discard(id(conn))
 self._connections.append(conn)
 
 @contextmanager
 def session(self):
 """Context manager for automatic acquire/release"""
 conn = self.get_connection()
 try:
 yield conn
 finally:
 self.release_connection(conn)
 
 def close_all(self):
 """Close all connections"""
 for conn in self._connections:
 conn.close()
 self._connections.clear()

# Global singleton
pool = ConnectionPool("sales.db", pool_size=20)

from database import pool

@app.get("/api/sales/region-summary")
def get_region_summary():
 with pool.session() as con:
 df = con.execute("""
 SELECT region, 
 ROUND(SUM(amount * quantity), 2) AS total_amount,
 COUNT(DISTINCT order_id) AS order_count
 FROM read_parquet('data/sales.parquet')
 GROUP BY region
 ORDER BY total_amount DESC
 """).df()
 return df.to_dict(orient="records")

import time
import json
import hashlib

class MultiLevelCache:
 """
 Multi-level cache: L1 Memory -> L2 Redis -> DuckDB direct query
 
 Use cases:
 - L1: Hot data (Dashboard homepage, TOP rankings), TTL 60s
 - L2: General query results (regional summary, trends), TTL 300s
 - Miss: Direct DuckDB query
 """
 
 def __init__(self, redis_url=None):
 self.l1_cache = {}
 self.l1_ttl = 60
 self.l2_ttl = 300
 
 if redis_url:
 try:
 import redis
 self.redis = redis.Redis.from_url(redis_url)
 self.has_l2 = True
 except ImportError:
 self.has_l2 = False
 else:
 self.has_l2 = False
 
 def _make_key(self, query, params):
 raw = query + ":" + json.dumps(params, sort_keys=True)
 return hashlib.md5(raw.encode()).hexdigest()
 
 def get(self, key):
 """Query multi-level cache"""
 if key in self.l1_cache:
 entry = self.l1_cache[key]
 if time.time() - entry['timestamp'] < self.l1_ttl:
 return entry['data']
 del self.l1_cache[key]
 
 if self.has_l2:
 try:
 data = self.redis.get(key)
 if data:
 result = json.loads(data)
 self.l1_cache[key] = {
 'data': result,
 'timestamp': time.time()
 }
 return result
 except Exception:
 pass
 
 return None
 
 def set(self, key, data, ttl=None):
 """Write to multi-level cache"""
 ttl = ttl or self.l1_ttl
 self.l1_cache[key] = {
 'data': data,
 'timestamp': time.time()
 }
 if self.has_l2 and ttl > self.l1_ttl:
 try:
 self.redis.setex(key, ttl, json.dumps(data))
 except Exception:
 pass
 
 def invalidate(self, pattern=None):
 """Invalidate cache entries"""
 if pattern:
 keys_to_del = [k for k in self.l1_cache if pattern in k]
 for k in keys_to_del:
 del self.l1_cache[k]
 if self.has_l2:
 keys = self.redis.keys(pattern + "*")
 if keys:
 self.redis.delete(*keys)
 else:
 self.l1_cache.clear()
 if self.has_l2:
 self.redis.flushdb()

cache = MultiLevelCache(redis_url="redis://localhost:6379/0")

@app.get("/api/sales/region-summary")
def get_region_summary():
 cache_key = cache._make_key("region_summary", {})
 
 cached = cache.get(cache_key)
 if cached is not None:
 return cached
 
 with pool.session() as con:
 df = con.execute("""
 SELECT region, 
 ROUND(SUM(amount * quantity), 2) AS total_amount,
 SUM(quantity) AS total_quantity,
 ROUND(AVG(amount), 2) AS avg_order_value,
 COUNT(DISTINCT order_id) AS order_count
 FROM read_parquet('data/sales.parquet')
 GROUP BY region
 ORDER BY total_amount DESC
 """).df()
 
 result = df.to_dict(orient="records")
 cache.set(cache_key, result, ttl=300)
 return result

from prometheus_client import Counter, Histogram

cache_hits = Counter('cache_hits_total', 'Cache hit count')
cache_misses = Counter('cache_misses_total', 'Cache miss count')
query_duration = Histogram('query_duration_seconds', 'Query execution time')

@app.get("/api/sales/daily-trend")
@query_duration.time()
def get_daily_trend(start_date="2025-01-01", end_date="2025-12-31"):
 cache_key = "daily_trend:" + start_date + ":" + end_date
 
 cached = cache.get(cache_key)
 if cached is not None:
 cache_hits.inc()
 return cached
 
 cache_misses.inc()
 
 with pool.session() as con:
 df = con.execute(f"""
 SELECT DATE(sale_date)::VARCHAR AS date,
 ROUND(SUM(amount * quantity), 2) AS revenue,
 COUNT(DISTINCT order_id) AS orders
 FROM read_parquet('data/sales.parquet')
 WHERE sale_date BETWEEN '{start_date}' AND '{end_date}'
 GROUP BY DATE(sale_date)
 ORDER BY date
 """).df()
 
 result = df.to_dict(orient="records")
 cache.set(cache_key, result, ttl=300)
 return result

from fastapi import FastAPI

app = FastAPI(title="DuckDB Data Product API", version="2.0.0")

@app.get("/health")
def health_check():
 status = {
 "status": "healthy",
 "timestamp": time.time(),
 "pool": {
 "available": len(pool._connections),
 "in_use": len(pool._in_use),
 "total": pool.pool_size
 },
 "cache": {
 "l1_size": len(cache.l1_cache),
 "l2_available": cache.has_l2
 }
 }
 return status

@app.get("/metrics")
def metrics():
 from prometheus_client import generate_latest
 return generate_latest()

import logging
import time

logger = logging.getLogger("slow_query")

def log_slow_query(operation, duration, threshold=1.0):
 if duration > threshold:
 logger.warning(
 "SLOW_QUERY: operation=%s, duration=%.3fs",
 operation, duration
 )

@app.middleware("http")
async def timing_middleware(request, call_next):
 start_time = time.time()
 response = await call_next(request)
 duration = time.time() - start_time
 
 if duration > 0.5:
 log_slow_query(request.url.path, duration)
 
 response.headers["X-Response-Time"] = "%.3fs" % duration
 return response

FROM python:3.11-slim

RUN apt-get update && apt-get install -y \
 gcc libpq-dev default-libmysqlclient-dev \
 && rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
RUN mkdir -p /app/data

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

version: '3.8'

services:
 app:
 build: .
 ports:
 - "8000:8000"
 environment:
 - REDIS_URL=redis://redis:6379/0
 - DB_PATH=/app/data/sales.db
 volumes:
 - ./data:/app/data
 depends_on:
 - redis
 deploy:
 resources:
 limits:
 memory: 2G
 cpus: '1.0'
 
 redis:
 image: redis:7-alpine
 ports:
 - "6379:6379"
 volumes:
 - redis_data:/data

volumes:
 redis_data:

#!/bin/bash
echo "Deploying DuckDB Data Product..."
git pull origin main
docker compose build --no-cache
docker compose down
docker compose up -d
sleep 5
curl -f http://localhost:8000/health || {
 echo "Health check failed!"
 docker compose logs app
 exit 1
}
echo "Deployment successful!"