Sync from GitHub main

app/cache.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, Optional, Tuple
+
+from prometheus_client import Counter
+
+cache_hits_total = Counter("nl2sql_cache_hits", "NL2SQL cache hits")
+cache_misses_total = Counter("nl2sql_cache_misses", "NL2SQL cache misses")
+
+
+class NL2SQLCache:
+    """
+    Tiny in-memory TTL cache for NL2SQL responses.
+    Stores serialized response payloads (dicts) keyed by a hash.
+    """
+
+    def __init__(self, ttl: float = 15.0) -> None:
+        self.ttl = ttl
+        self._store: Dict[str, Tuple[float, Dict[str, Any]]] = {}
+
+    def _gc(self, now: float) -> None:
+        """Remove expired entries based on the configured TTL."""
+        expired_keys = [
+            key for key, (ts, _) in self._store.items() if now - ts > self.ttl
+        ]
+        for key in expired_keys:
+            del self._store[key]
+
+    def get(self, key: str) -> Optional[Dict[str, Any]]:
+        """
+        Return cached payload if present and not expired, otherwise None.
+        Also updates Prometheus counters for hits/misses.
+        """
+        now = time.time()
+        self._gc(now)
+
+        entry = self._store.get(key)
+        if entry is None:
+            cache_misses_total.inc()
+            return None
+
+        ts, payload = entry
+        if now - ts <= self.ttl:
+            cache_hits_total.inc()
+            return payload
+
+        # Entry is expired
+        del self._store[key]
+        cache_misses_total.inc()
+        return None
+
+    def set(self, key: str, payload: Dict[str, Any]) -> None:
+        """Store payload under the given key with current timestamp."""
+        self._store[key] = (time.time(), payload)

app/dependencies.py

@@ -0,0 +1,25 @@
+from functools import lru_cache
+
+from app.services.nl2sql_service import NL2SQLService
+from app.cache import NL2SQLCache
+from app.settings import get_settings
+
+
+@lru_cache()
+def get_nl2sql_service() -> NL2SQLService:
+    """
+    Singleton-ish NL2SQLService for the FastAPI app.
+
+    Uses centralized Settings so configuration is loaded once and injected.
+    """
+    settings = get_settings()
+    return NL2SQLService(settings=settings)
+
+
+@lru_cache()
+def get_cache() -> NL2SQLCache:
+    """
+    Singleton in-memory cache for NL2SQL responses.
+    TTL is intentionally short; this is a per-process best-effort cache.
+    """
+    return NL2SQLCache(ttl=15.0)

app/errors.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class AppError(Exception):
+    """Base class for domain-level errors."""
+
+    message: str
+
+    def __str__(self) -> str:
+        return self.message
+
+
+# 4xx-ish
+@dataclass
+class DbNotFound(AppError):
+    """Requested DB (or db_id) does not exist."""
+
+
+@dataclass
+class InvalidRequest(AppError):
+    """User input is invalid or cannot be processed."""
+
+
+@dataclass
+class SchemaRequired(AppError):
+    """Caller must provide schema_preview (e.g. postgres mode)."""
+
+
+@dataclass
+class SchemaDeriveError(AppError):
+    """Failed to derive schema preview from DB."""
+
+
+# 5xx-ish
+@dataclass
+class PipelineConfigError(AppError):
+    """Pipeline/YAML/config is missing or malformed."""
+
+
+@dataclass
+class PipelineRunError(AppError):
+    """Unexpected failure while running the pipeline."""

app/exception_handlers.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from app.errors import AppError
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+    """
+    Register global exception handlers for the FastAPI application.
+    """
+
+    @app.exception_handler(AppError)
+    async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
+        """
+        Map domain-level AppError instances to HTTP responses.
+        This keeps routers thin and lets the domain raise AppError freely.
+        """
+        status = getattr(exc, "http_status", 500)
+        code = getattr(exc, "code", "app_error")
+        message = getattr(exc, "message", str(exc))
+        extra: Dict[str, Any] = getattr(exc, "extra", {}) or {}
+
+        payload = {
+            "code": code,
+            "message": message,
+            "extra": extra,
+        }
+        return JSONResponse(status_code=status, content=payload)

app/main.py

@@ -1,28 +1,35 @@
 import os
 import time
+
 from fastapi import FastAPI, Request, Response, HTTPException
 from fastapi.responses import PlainTextResponse, RedirectResponse
 from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
+
 from nl2sql.prom import REGISTRY
-from app.routers import dev
+from app.routers import dev, nl2sql
+from app.settings import get_settings
+from app.exception_handlers import register_exception_handlers
 
 try:
     from dotenv import load_dotenv
 
     load_dotenv()
 except Exception:
+    # Best-effort .env loading; app must not crash if dotenv is missing.
     pass
 
-from app.routers import nl2sql
+
+settings = get_settings()
 
 # ----------------------------------------------------------------------------
 #  App definition
 # ----------------------------------------------------------------------------
 application = FastAPI(
     title="NL2SQL Copilot Prototype",
-    version=os.getenv("APP_VERSION", "0.1.0"),
+    version=settings.app_version,
     description="Convert natural language to safe & verified SQL",
 )
+register_exception_handlers(application)
 
 # Register only versioned API
 application.include_router(nl2sql.router, prefix="/api/v1")
@@ -30,6 +37,7 @@ application.include_router(nl2sql.router, prefix="/api/v1")
 # Register Dev-only routes (only when APP_ENV=dev)
 if os.getenv("APP_ENV", "dev").lower() == "dev":
     application.include_router(dev.router, prefix="/api/v1")
+
 # ----------------------------------------------------------------------------
 #  Prometheus Metrics Middleware
 # ----------------------------------------------------------------------------
@@ -75,21 +83,30 @@ def healthz() -> str:
 
 @application.get("/readyz", response_class=PlainTextResponse, tags=["system"])
 def readyz() -> str:
-    mode = os.getenv("DB_MODE", "sqlite").lower()
+    """
+    Lightweight readiness probe:
+
+    - For postgres mode → ping PostgresAdapter using configured DSN.
+    - For sqlite mode   → ping SQLiteAdapter using configured default path.
+    """
+    mode = settings.db_mode.lower()
     try:
         if mode == "postgres":
             from adapters.db.postgres_adapter import PostgresAdapter
 
-            pg = PostgresAdapter(os.environ["POSTGRES_DSN"])
+            dsn = (settings.postgres_dsn or "").strip()
+            if not dsn:
+                raise RuntimeError("POSTGRES_DSN is not configured for readiness check")
+
+            pg = PostgresAdapter(dsn)
             ping_fn = getattr(pg, "ping", None)
             if callable(ping_fn):
                 ping_fn()
         else:
             from adapters.db.sqlite_adapter import SQLiteAdapter
 
-            sq = SQLiteAdapter(
-                os.getenv("DEFAULT_SQLITE_PATH", "data/Chinook_Sqlite.sqlite")
-            )
+            db_path = settings.default_sqlite_path or "data/Chinook_Sqlite.sqlite"
+            sq = SQLiteAdapter(db_path)
             ping_fn = getattr(sq, "ping", None)
             if callable(ping_fn):
                 ping_fn()
@@ -105,6 +122,7 @@ def root():
 
 @application.get("/health")
 def health():
+    # This is a higher-level health stub; real checks can be wired later
     return {"status": "ok", "db": "connected", "llm": "reachable", "uptime_sec": 123.4}
 
 

app/routers/nl2sql.py

@@ -4,82 +4,51 @@ from __future__ import annotations
 from dataclasses import asdict, is_dataclass
 import os
 from pathlib import Path
-import time
 import uuid
-from typing import Any, Dict, Optional, Union, cast, Callable, Tuple
+from typing import Any, Dict, Optional, Tuple, cast
 import hashlib
 import logging
 
 # --- Third-party ---
-from fastapi import APIRouter, HTTPException, UploadFile, File, Depends, Query
-from fastapi import Security
+from fastapi import APIRouter, Depends, HTTPException, Security, UploadFile, File
 from fastapi.security import APIKeyHeader
 from prometheus_client import Counter
 
 # --- Local ---
 from app.schemas import NL2SQLRequest, NL2SQLResponse, ClarifyResponse
-from app.state import cleanup_stale_dbs, register_db
-from nl2sql.pipeline import FinalResult, FinalResult as _FinalResult
-from adapters.llm.openai_provider import OpenAIProvider
-from adapters.db.sqlite_adapter import SQLiteAdapter
-from adapters.db.postgres_adapter import PostgresAdapter
-from nl2sql.pipeline_factory import (
-    pipeline_from_config_with_adapter,
-)
+from app.state import register_db
+from nl2sql.pipeline import FinalResult
 from nl2sql.prom import REGISTRY
+from app.dependencies import get_cache, get_nl2sql_service
+from app.cache import NL2SQLCache
+from app.services.nl2sql_service import NL2SQLService
+from app.settings import get_settings
+from app.errors import (
+    AppError,
+)
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
 
-log = logging.getLogger(__name__)
 api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
 
 
 def require_api_key(key: Optional[str] = Security(api_key_header)):
-    raw = os.getenv("API_KEYS", "")
+    """
+    Simple API key check using X-API-Key header and configured API keys.
+
+    - Settings.api_keys_raw is a comma-separated list of keys.
+    - If api_keys_raw is empty → auth disabled (dev mode).
+    """
+    raw = settings.api_keys_raw or ""
     allowed = {k.strip() for k in raw.split(",") if k.strip()}
-    if not allowed:  # no keys set → auth disabled (dev mode)
+    if not allowed:
+        # No keys configured → treat as dev mode (auth off).
         return
     if not key or key not in allowed:
         raise HTTPException(status_code=401, detail="invalid API key")
 
 
-_PIPELINE: Optional[Any] = None  # lazy cache
-
-
-Runner = Callable[..., _FinalResult]
-
-
-def get_runner() -> Runner:
-    """Build pipeline lazily; under pytest return a stub runner."""
-    if os.getenv("PYTEST_CURRENT_TEST"):
-        # Minimal OK runner for route tests (no ambiguity)
-        def _fake_runner(
-            *, user_query: str, schema_preview: str | None = None
-        ) -> _FinalResult:
-            return _FinalResult(
-                ok=True,
-                ambiguous=False,
-                error=False,
-                details=None,
-                questions=None,
-                sql="SELECT 1;",
-                rationale=None,
-                verified=True,
-                traces=[],
-            )
-
-        return _fake_runner
-
-    global _PIPELINE
-    if _PIPELINE is None:
-        adapter = _select_adapter(None)  # fallback demo.db
-        _PIPELINE = pipeline_from_config_with_adapter(CONFIG_PATH, adapter=adapter)
-    return _PIPELINE.run
-
-
-def _build_pipeline(adapter) -> Any:
-    """Thin wrapper for tests to monkeypatch; builds a pipeline bound to adapter."""
-    return pipeline_from_config_with_adapter(CONFIG_PATH, adapter=adapter)
-
-
 ####################################
 # ---- Simple in-memory cache for NL→SQL responses ----
 
@@ -87,29 +56,54 @@ cache_hits_total = Counter("cache_hits_total", "NL2SQL cache hits", registry=REG
 cache_misses_total = Counter(
     "cache_misses_total", "NL2SQL cache misses", registry=REGISTRY
 )
-_CACHE_TTL = int(os.getenv("NL2SQL_CACHE_TTL_SEC", "300"))  # 5 minutes
-_CACHE_MAX = int(os.getenv("NL2SQL_CACHE_MAX", "256"))
+
+# Cache TTL and max size from centralized settings
+_CACHE_TTL = settings.cache_ttl_sec
+_CACHE_MAX = settings.cache_max_entries
 _CACHE: Dict[Tuple[str, str, str], Tuple[float, Dict[str, Any]]] = {}
 
 
 def _norm_q(s: str) -> str:
+    """Normalize a user query for cache key purposes."""
     return (s or "").strip().lower()
 
 
 def _schema_key(preview: str) -> str:
+    """Hash the schema preview so we do not store huge strings in the cache key."""
     return hashlib.md5((preview or "").encode()).hexdigest()
 
 
-def _ck(db_id: Optional[str], query: str, preview: str) -> Tuple[str, str, str]:
-    return (db_id or "default", _norm_q(query), _schema_key(preview))
+def _ck(
+    db_id: Optional[str],
+    query: str,
+    schema_preview: str,
+) -> str:
+    """
+    Build a stable cache key for (db_id, query, schema_preview).
+
+    We keep the external cache API string-based, and hash the
+    potentially large schema_preview to avoid huge dictionary keys.
+    """
+    # Normalize db_id
+    db_part = db_id or "__default__"
+
+    # Build a single string seed
+    seed = f"{db_part}\n{query}\n{schema_preview}"
+
+    # Short, deterministic key
+    return hashlib.sha1(seed.encode("utf-8")).hexdigest()
 
 
 def _cache_gc(now: float) -> None:
+    """
+    Garbage-collect cache entries by TTL and max size.
+    """
     # TTL eviction
     for k, (ts, _) in list(_CACHE.items()):
         if now - ts > _CACHE_TTL:
             _CACHE.pop(k, None)
-    # size eviction
+
+    # Size eviction (naive FIFO-style)
     while len(_CACHE) > _CACHE_MAX:
         _CACHE.pop(next(iter(_CACHE)), None)
 
@@ -121,92 +115,20 @@ router = APIRouter(prefix="/nl2sql")
 # -------------------------------
 # Config / Defaults
 # -------------------------------
-DB_MODE = os.getenv("DB_MODE", "sqlite").lower()  # "sqlite" or "postgres"
-POSTGRES_DSN = os.getenv("POSTGRES_DSN")
-# Default demo DB used when no db_id is provided (can be full Chinook or a tiny demo DB)
-DEFAULT_SQLITE_PATH: str = os.getenv(
-    "DEFAULT_SQLITE_PATH", "data/Chinook_Sqlite.sqlite"
-)
-print("=== STARTUP DEBUG ===")
-print("DEFAULT_SQLITE_PATH:", DEFAULT_SQLITE_PATH)
-print("CWD:", os.getcwd())
-print("FILES in ./:", os.listdir("."))
-print(
-    "FILES in ./data:",
-    os.listdir("data") if os.path.exists("data") else "NO DATA FOLDER",
-)
-
+DB_MODE = settings.db_mode.lower()  # "sqlite" or "postgres"
 
-# Runtime upload storage
-_DB_UPLOAD_DIR = os.getenv("DB_UPLOAD_DIR", "/tmp/nl2sql_dbs")
-_DB_TTL_SECONDS: int = int(os.getenv("DB_TTL_SECONDS", "7200"))  # default 2 hours
+# Runtime upload storage for SQLite DBs
+_DB_UPLOAD_DIR = settings.db_upload_dir
 os.makedirs(_DB_UPLOAD_DIR, exist_ok=True)
 
-# Persisted map
-_DB_MAP_PATH = Path("data/uploads/db_map.json")
-_DB_MAP_PATH.parent.mkdir(parents=True, exist_ok=True)
-
+# Optional: separate directory for other uploads (kept as-is for now)
 UPLOAD_DIR = Path("data/uploads")
 UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
 
-CONFIG_PATH = os.getenv("PIPELINE_CONFIG", "configs/sqlite_pipeline.yaml")
-
-
-# -------------------------------
-# Adapter selection (lazy)
-# -------------------------------
-def _select_adapter(db_id: Optional[str]) -> Union[PostgresAdapter, SQLiteAdapter]:
-    """
-    Resolve DB adapter path for SQLite or Postgres.
-    """
-    if DB_MODE == "postgres":
-        dsn = os.environ.get("POSTGRES_DSN")
-        if not dsn:
-            raise HTTPException(status_code=500, detail="POSTGRES_DSN env is missing")
-        return PostgresAdapter(dsn)
-
-    if db_id:
-        cleanup_stale_dbs()
-
-        candidates = [
-            Path("/tmp/nl2sql_dbs") / f"{db_id}.sqlite",
-            Path("/tmp/nl2sql_dbs") / f"{db_id}.db",
-            Path("data/uploads") / f"{db_id}.sqlite",
-            Path("data/uploads") / f"{db_id}.db",
-            Path("data") / f"{db_id}.sqlite",
-            Path("data") / f"{db_id}.db",
-        ]
-
-        for candidate in candidates:
-            if candidate.exists():
-                log.info(f"Using DB file: {candidate}")
-                return SQLiteAdapter(str(candidate))
-
-        raise HTTPException(status_code=404, detail=f"db_id not found: {db_id}")
-
-    # -------- Default SQLite Logic --------
-    default_path = Path(DEFAULT_SQLITE_PATH)
-    db_path = str(default_path)
-
-    log.debug("DEFAULT SQLITE DEBUG INFO:")
-    log.debug(f"DEFAULT_SQLITE_PATH env: {DEFAULT_SQLITE_PATH}")
-    log.debug(f"CWD: {os.getcwd()}")
-    log.debug(f"ABS PATH: {default_path.resolve()}")
-    log.debug(f"EXISTS?: {default_path.exists()}")
-    if os.path.exists("data"):
-        log.debug(f"LIST DATA: {os.listdir('data')}")
-    else:
-        log.debug("LIST DATA: NO DATA DIRECTORY")
-
-    if not default_path.exists():
-        fallback = Path("data/demo.db")
-        if fallback.exists():
-            log.warning("Default sqlite missing; using fallback demo.db")
-            db_path = str(fallback)
-        else:
-            raise HTTPException(status_code=500, detail="no sqlite database found")
-
-    return SQLiteAdapter(db_path)
+logger.debug(
+    "NL2SQL router configured",
+    extra={"db_mode": DB_MODE, "upload_dir": _DB_UPLOAD_DIR},
+)
 
 
 # -------------------------------
@@ -215,37 +137,43 @@ def _select_adapter(db_id: Optional[str]) -> Union[PostgresAdapter, SQLiteAdapte
 
 
 @router.get("/schema")
-def get_schema(db_id: Optional[str] = Query(default=None)):
+def schema_endpoint(
+    db_id: Optional[str] = None,
+    svc: NL2SQLService = Depends(get_nl2sql_service),
+):
     """
-    Return a schema preview for a given db_id (SQLite only).
-    If db_id is omitted, returns the default database schema.
+    Return a lightweight schema preview string for the given DB.
+
+    - If db_id is provided, service will resolve the uploaded DB.
+    - If not, service falls back to the default DB.
+    - In postgres mode, caller must usually provide schema_preview explicitly.
+    Domain errors (AppError subclasses) are handled by the global exception handler.
+    This endpoint only wraps truly unexpected errors into a generic HTTP 500
     """
     try:
-        adapter = _select_adapter(db_id)
-        preview = _derive_schema_preview(adapter)
-        if not preview.strip():
-            raise HTTPException(
-                status_code=404, detail="Schema preview not available or empty"
-            )
-        return {"db_id": db_id or "default", "schema_preview": preview}
-    except HTTPException:
+        preview = svc.get_schema_preview(db_id=db_id, override=None)
+    except AppError:
+        # Let the global AppError handler deal with it.
         raise
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Schema introspection failed: {e}")
+    except Exception as exc:
+        logger.exception("Unexpected error in schema_endpoint", exc_info=exc)
+        raise HTTPException(
+            status_code=500,
+            detail="failed to derive schema preview",
+        ) from exc
 
-
-# -------------------------------
-# LLM & Pipeline builders (lazy)
-# -------------------------------
-def _get_llm() -> OpenAIProvider:
-    # Create provider on demand, after .env has been loaded in app.main
-    return OpenAIProvider()
+    return {"schema_preview": preview}
 
 
 # -------------------------------
 # Helpers
 # -------------------------------
+
+
 def _to_dict(obj: Any) -> Any:
+    """
+    Convert dataclass-like objects (and similar) to plain dicts for JSON.
+    """
     if is_dataclass(obj) and not isinstance(obj, type):
         return asdict(obj)  # type: ignore[arg-type]
     return obj
@@ -254,6 +182,7 @@ def _to_dict(obj: Any) -> Any:
 def _round_trace(t: Any) -> Dict[str, Any]:
     """
     Normalize a trace entry (dict or StageTrace-like object) for API/UI:
+
     - stage: str (required)
     - duration_ms: int (rounded)
     - summary: optional (pass-through if exists)
@@ -277,7 +206,7 @@ def _round_trace(t: Any) -> Dict[str, Any]:
         token_in = getattr(t, "token_in", None)
         token_out = getattr(t, "token_out", None)
 
-    # coerce duration to int with rounding
+    # Coerce duration to int with rounding
     try:
         ms_int = int(round(float(ms))) if ms is not None else 0
     except Exception:
@@ -301,8 +230,17 @@ def _round_trace(t: Any) -> Dict[str, Any]:
 # -------------------------------
 # Upload endpoint (SQLite only)
 # -------------------------------
+
+
 @router.post("/upload_db", dependencies=[Depends(require_api_key)])
 async def upload_db(file: UploadFile = File(...)):
+    """
+    Upload a SQLite DB file and register it under a generated db_id.
+
+    Only available when DB_MODE is 'sqlite':
+    - Allowed extensions: .db, .sqlite
+    - File size capped by configured upload_max_bytes (default 20 MB)
+    """
     if DB_MODE != "sqlite":
         raise HTTPException(
             status_code=400, detail="DB upload is only supported in sqlite mode"
@@ -315,7 +253,7 @@ async def upload_db(file: UploadFile = File(...)):
         )
 
     data = await file.read()
-    max_bytes = int(os.getenv("UPLOAD_MAX_BYTES", str(20 * 1024 * 1024)))  # 20 MB
+    max_bytes = settings.upload_max_bytes
     if len(data) > max_bytes:
         raise HTTPException(
             status_code=400, detail=f"File too large (> {max_bytes} bytes)"
@@ -327,92 +265,119 @@ async def upload_db(file: UploadFile = File(...)):
         with open(out_path, "wb") as f:
             f.write(data)
     except Exception as e:
+        logger.debug("Failed to store uploaded DB file", exc_info=e)
         raise HTTPException(status_code=500, detail=f"Failed to store DB: {e}")
 
     register_db(db_id, out_path)
+    logger.debug("Registered uploaded DB", extra={"db_id": db_id, "path": out_path})
     return {"db_id": db_id}
 
 
-def _final_schema_preview(db_id: Optional[str], provided_preview: Optional[str]) -> str:
-    if provided_preview and provided_preview.strip():
-        return provided_preview
-
-    adapter = _select_adapter(db_id)  # works for both None and explicit db_id
-    return _derive_schema_preview(adapter) or ""
-
-
 @router.get("/health")
 def health():
-    return {"status": "ok", "version": os.getenv("APP_VERSION", "dev")}
+    """Simple router-level health endpoint."""
+    return {"status": "ok", "version": settings.app_version}
 
 
 # -------------------------------
 # Main NL2SQL endpoint
 # -------------------------------
+
+
 @router.post("", name="nl2sql_handler", dependencies=[Depends(require_api_key)])
 def nl2sql_handler(
     request: NL2SQLRequest,
-    run: Runner = Depends(get_runner),
-):
+    svc: NL2SQLService = Depends(get_nl2sql_service),
+    cache: NL2SQLCache = Depends(get_cache),
+) -> NL2SQLResponse | ClarifyResponse | Dict[str, Any]:
     """
-    NL→SQL handler using YAML-driven DI. If 'db_id' is provided, we override only the adapter
-    while keeping all other stages from the YAML configs intact.
+    Main NL→SQL handler.
+
+    Flow:
+    - Resolve schema preview (client override or derived from DB).
+    - Check in-memory cache (db_id + query + schema hash).
+    - Run the pipeline through NL2SQLService.
+    - Map FinalResult to API response or HTTP error.
     """
     db_id = getattr(request, "db_id", None)
-    final_preview = _final_schema_preview(
-        db_id, cast(Optional[str], getattr(request, "schema_preview", None))
-    )
+
+    # ---- schema preview ----
+    try:
+        final_preview = svc.get_schema_preview(
+            db_id=db_id,
+            override=request.schema_preview,
+        )
+    except AppError:
+        # Domain-level errors are handled by the global AppError handler.
+        raise
+    except Exception as exc:
+        logger.exception(
+            "Unexpected error while preparing schema preview",
+            exc_info=exc,
+        )
+        raise HTTPException(
+            status_code=500,
+            detail="failed to prepare schema",
+        ) from exc
 
     # ---- cache lookup ----
-    now = time.time()
-    _cache_gc(now)
-    ck = _ck(db_id, request.query, final_preview)
-    hit = _CACHE.get(ck)
-    if hit and now - hit[0] <= _CACHE_TTL:
-        cache_hits_total.inc()
-        return hit[1]  # early return
-    cache_misses_total.inc()
-
-    # Choose runner: default pipeline from YAML OR per-request override with a specific adapter
-    if db_id:
-        adapter = _select_adapter(db_id)
-        pipeline = _build_pipeline(adapter)
-        runner = pipeline.run
-    else:
-        runner = run
+    cache_key = _ck(db_id, request.query, final_preview)
+    cached_payload = cache.get(cache_key)
+    if cached_payload is not None:
+        return cached_payload
 
-    # Execute pipeline
+    # ---- pipeline execution via service ----
     try:
-        result = runner(user_query=request.query, schema_preview=final_preview)
+        result = svc.run_query(
+            query=request.query,
+            db_id=db_id,
+            schema_preview=final_preview,
+        )
+    except AppError:
+        # Let the global handler convert it to an HTTP response.
+        raise
     except Exception as exc:
-        raise HTTPException(status_code=500, detail=f"Pipeline crash: {exc!s}")
+        logger.exception(
+            "Unexpected pipeline crash in NL2SQLService.run_query",
+            exc_info=exc,
+        )
+        raise HTTPException(
+            status_code=500,
+            detail="internal pipeline error",
+        ) from exc
 
-    # Type sanity
+    # ---- type sanity check ----
     if not isinstance(result, FinalResult):
-        raise HTTPException(status_code=500, detail="Pipeline returned unexpected type")
+        logger.debug(
+            "Pipeline returned unexpected type",
+            extra={"type": type(result).__name__},
+        )
+        raise HTTPException(
+            status_code=500,
+            detail="pipeline returned unexpected type",
+        )
 
-    # Ambiguity path → 200 with questions
+    # ---- ambiguity path → 200 with clarification questions ----
     if result.ambiguous:
         qs = result.questions or []
         return ClarifyResponse(ambiguous=True, questions=qs)
 
-    if not isinstance(result, _FinalResult):
-        raise HTTPException(status_code=500, detail="Pipeline returned unexpected type")
-
-    # Error path → 400 with joined details
+    # ---- error path → 400 with joined details ----
     if (not result.ok) or result.error:
-        print("❌ Pipeline failure dump:")
-        print("  ok:", result.ok)
-        print("  error:", result.error)
-        print("  details:", result.details)
-        print("  traces:", result.traces)
+        logger.debug(
+            "Pipeline reported failure",
+            extra={
+                "ok": result.ok,
+                "error": result.error,
+                "details": result.details,
+            },
+        )
         message = "; ".join(result.details or []) or "Unknown error"
         raise HTTPException(status_code=400, detail=message)
 
-    # Success path → 200 (coerce/standardize traces for API)
+    # ---- success path → 200 (normalize traces and executor result) ----
     traces = [_round_trace(t) for t in (result.traces or [])]
 
-    # Normalize execution result (if executor attached one)
     response_result: Dict[str, Any] = {}
     raw_result = getattr(result, "result", None)
     if raw_result is not None:
@@ -429,35 +394,6 @@ def nl2sql_handler(
         result=response_result,
     )
 
-    # store in cache
-    _CACHE[ck] = (time.time(), cast(Dict[str, Any], payload.model_dump()))
+    # Store in cache (as plain dict)
+    cache.set(cache_key, payload.model_dump())
     return payload
-
-
-def _derive_schema_preview(adapter: Union[PostgresAdapter, SQLiteAdapter]) -> str:
-    """
-    Build a strict, exact-cased schema preview for the LLM (SQLite only).
-    """
-    import sqlite3
-
-    db_path: Optional[str] = cast(
-        Optional[str], getattr(adapter, "db_path", None)
-    ) or cast(Optional[str], getattr(adapter, "path", None))
-    if not db_path or not os.path.exists(db_path):
-        return ""
-
-    try:
-        conn = sqlite3.connect(db_path)
-        cur = conn.cursor()
-        tables = cur.execute(
-            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
-        ).fetchall()
-        lines = []
-        for (tname,) in tables:
-            cols = cur.execute(f"PRAGMA table_info('{tname}')").fetchall()
-            colnames = [c[1] for c in cols]  # (cid, name, type, notnull, dflt, pk)
-            lines.append(f"{tname}({', '.join(colnames)})")
-        conn.close()
-        return "\n".join(lines)
-    except Exception:
-        return ""

app/services/nl2sql_service.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import sqlite3
+from dataclasses import dataclass
+from typing import Any, Optional
+from pathlib import Path
+
+from nl2sql.pipeline import FinalResult
+from nl2sql.pipeline_factory import pipeline_from_config_with_adapter
+from adapters.db.sqlite_adapter import SQLiteAdapter
+from adapters.db.postgres_adapter import PostgresAdapter
+from app import state
+from app.settings import Settings
+from app.errors import (
+    AppError,
+    DbNotFound,
+    SchemaRequired,
+    SchemaDeriveError,
+    PipelineConfigError,
+    PipelineRunError,
+)
+
+Adapter = Any  # You can replace this with a Protocol later
+
+
+@dataclass
+class NL2SQLService:
+    """
+    Application-level service for the NL2SQL use-case.
+
+    Responsibilities:
+        - Choose the right DB adapter based on db_mode + db_id.
+        - Derive or accept schema preview.
+        - Build and run the pipeline for a given query.
+    """
+
+    settings: Settings
+
+    def _select_adapter(self, db_id: Optional[str]) -> Adapter:
+        mode = self.settings.db_mode.lower()
+
+        if mode == "postgres":
+            dsn = (self.settings.postgres_dsn or "").strip()
+            if not dsn:
+                raise PipelineConfigError("Postgres DSN is not configured")
+            return PostgresAdapter(dsn=dsn)
+
+        if db_id:
+            state.cleanup_stale_dbs()
+            path = state.get_db_path(db_id)
+            if not path:
+                raise DbNotFound(f"Could not resolve DB for db_id={db_id!r}")
+            return SQLiteAdapter(path=path)
+
+        default_path = self.settings.default_sqlite_path
+        if not Path(default_path).exists():
+            raise DbNotFound(f"SQLite database path does not exist: {default_path!r}")
+
+        return SQLiteAdapter(path=default_path)
+
+    def _introspect_sqlite_schema(self, adapter: Adapter) -> str:
+        """
+        Build a lightweight textual schema preview for a SQLite database.
+
+        This is a straight port of the previous sqlite3 logic, but contained
+        inside the service instead of the router.
+        """
+        # Try to locate the underlying .db path from the adapter
+        db_path = getattr(adapter, "db_path", None) or getattr(adapter, "path", None)
+        if not db_path:
+            raise RuntimeError(
+                "SQLite adapter must expose a .db_path or .path attribute"
+            )
+
+        if not Path(db_path).exists():
+            raise FileNotFoundError(f"SQLite database path does not exist: {db_path}")
+
+        lines: list[str] = []
+        conn = sqlite3.connect(db_path)
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' "
+                "AND name NOT LIKE 'sqlite_%' ORDER BY name"
+            )
+            tables = [row[0] for row in cur.fetchall()]
+
+            for table in tables:
+                cur.execute(f"PRAGMA table_info({table})")
+                cols = [row[1] for row in cur.fetchall()]
+                if cols:
+                    lines.append(f"{table}({', '.join(cols)})")
+        finally:
+            conn.close()
+
+        return "\n".join(lines)
+
+    def get_schema_preview(
+        self,
+        db_id: Optional[str],
+        override: Optional[str],
+    ) -> str:
+        """
+        Decide which schema preview to use.
+
+        - If override is provided by the client → use it.
+        - Else, in sqlite mode → introspect the DB.
+        - In postgres mode without override → fail fast, the caller can map
+          this to a proper HTTP error.
+        """
+        if override:
+            return override
+
+        mode = self.settings.db_mode.lower()
+        if mode == "postgres":
+            raise SchemaRequired("schema_preview is required in postgres mode")
+
+        try:
+            adapter = self._select_adapter(db_id)
+            return self._introspect_sqlite_schema(adapter)
+        except DbNotFound:
+            raise
+        except Exception as exc:
+            raise SchemaDeriveError("failed to derive schema preview") from exc
+
+    def run_query(
+        self,
+        *,
+        query: str,
+        db_id: Optional[str],
+        schema_preview: str,
+    ) -> FinalResult:
+        """Build a pipeline for the given DB and run the query through it."""
+        try:
+            adapter = self._select_adapter(db_id)
+        except AppError:
+            raise
+        except Exception as exc:
+            raise PipelineRunError("failed to select adapter") from exc
+
+        try:
+            pipeline = pipeline_from_config_with_adapter(
+                self.settings.pipeline_config_path,
+                adapter=adapter,
+            )
+        except FileNotFoundError as exc:
+            raise PipelineConfigError(
+                f"Pipeline config not found at {self.settings.pipeline_config_path!r}"
+            ) from exc
+        except Exception as exc:
+            raise PipelineConfigError(
+                f"Failed to build pipeline from {self.settings.pipeline_config_path!r}: {exc}"
+            ) from exc
+
+        try:
+            result = pipeline.run(user_query=query, schema_preview=schema_preview)
+        except AppError:
+            raise
+        except Exception as exc:
+            raise PipelineRunError("pipeline crashed during execution") from exc
+
+        return result

app/settings.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+
+# Resolve repo root from this file's location:
+# app/settings.py → parent = app/ → parent = repo root
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+# Canonical demo DB and pipeline config shipped with the repo
+DEFAULT_DEMO_DB = REPO_ROOT / "data" / "demo.db"
+DEFAULT_PIPELINE_CONFIG = REPO_ROOT / "configs" / "sqlite_pipeline.yaml"
+
+
+@dataclass
+class Settings:
+    """
+    Centralized application configuration.
+
+    Does NOT depend on pydantic. Values are loaded from environment
+    variables via Settings.from_env().
+    """
+
+    # --- DB mode / adapters ---
+    db_mode: str = "sqlite"  # "sqlite" or "postgres"
+    postgres_dsn: str = ""
+
+    # --- Pipeline config ---
+    pipeline_config_path: str = str(DEFAULT_PIPELINE_CONFIG)
+
+    # --- SQLite uploaded DBs ---
+    db_upload_dir: str = "/tmp/nl2sql_dbs"
+    db_ttl_seconds: int = 7200  # 2 hours
+
+    # --- Upload constraints ---
+    upload_max_bytes: int = 20 * 1024 * 1024  # 20MB
+
+    # --- Cache settings ---
+    cache_ttl_sec: int = 300
+    cache_max_entries: int = 256
+
+    # --- Default SQLite path (demo DB) ---
+    default_sqlite_path: str = str(DEFAULT_DEMO_DB)
+
+    # --- API keys (comma-separated) ---
+    api_keys_raw: str = ""
+
+    # --- App version ---
+    app_version: str = "dev"
+
+    @classmethod
+    def from_env(cls) -> "Settings":
+        """
+        Build Settings from environment variables with sane fallbacks.
+
+        - DEFAULT_SQLITE_PATH and PIPELINE_CONFIG can be absolute or relative.
+        - Relative paths are resolved against REPO_ROOT.
+        """
+
+        def getenv_int(name: str, default: int) -> int:
+            raw = os.getenv(name)
+            if raw is None or raw.strip() == "":
+                return default
+            try:
+                return int(raw)
+            except ValueError:
+                return default
+
+        # --- Default SQLite path ---
+        raw_default_db = os.getenv("DEFAULT_SQLITE_PATH", "").strip()
+        if raw_default_db:
+            db_candidate = Path(raw_default_db)
+            if not db_candidate.is_absolute():
+                db_candidate = REPO_ROOT / raw_default_db
+        else:
+            db_candidate = DEFAULT_DEMO_DB
+
+        # --- Pipeline config path ---
+        raw_cfg = os.getenv("PIPELINE_CONFIG", "").strip()
+        if raw_cfg:
+            cfg_candidate = Path(raw_cfg)
+            if not cfg_candidate.is_absolute():
+                cfg_candidate = REPO_ROOT / raw_cfg
+        else:
+            cfg_candidate = DEFAULT_PIPELINE_CONFIG
+
+        return cls(
+            db_mode=os.getenv("DB_MODE", cls.db_mode),
+            postgres_dsn=os.getenv("POSTGRES_DSN", cls.postgres_dsn),
+            pipeline_config_path=str(cfg_candidate),
+            db_upload_dir=os.getenv("DB_UPLOAD_DIR", cls.db_upload_dir),
+            db_ttl_seconds=getenv_int("DB_TTL_SECONDS", cls.db_ttl_seconds),
+            upload_max_bytes=getenv_int("UPLOAD_MAX_BYTES", cls.upload_max_bytes),
+            cache_ttl_sec=getenv_int("NL2SQL_CACHE_TTL_SEC", cls.cache_ttl_sec),
+            cache_max_entries=getenv_int("NL2SQL_CACHE_MAX", cls.cache_max_entries),
+            default_sqlite_path=str(db_candidate),
+            api_keys_raw=os.getenv("API_KEYS", cls.api_keys_raw),
+            app_version=os.getenv("APP_VERSION", cls.app_version),
+        )
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    return Settings.from_env()

app/state.py

@@ -1,79 +1,166 @@
+from __future__ import annotations
+
+import logging
 import os
 import time
-import logging
 from pathlib import Path
-from typing import Optional, TypedDict
+from typing import Dict, Tuple, Optional
 
 log = logging.getLogger(__name__)
 
-# ------------------------------
-# Config
-# ------------------------------
-
-# default upload directory (can override via .env)
-_DB_UPLOAD_DIR = Path(os.getenv("DB_UPLOAD_DIR", "/tmp/nl2sql_dbs"))
-_DB_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
-
-
-class DBEntry(TypedDict):
-    path: str
-    ts: float
-
 
-# in-memory map: {db_id: {"path": str, "ts": float}}
-DB_MAP: dict[str, DBEntry] = {}
-
-# cleanup threshold (hours)
-DB_TTL_HOURS = 6
-
-
-# ------------------------------
-# Helpers
-# ------------------------------
+class DbUploadStore:
+    """
+    In-memory registry for uploaded DB files with simple TTL-based cleanup.
+
+    Responsibilities:
+    - Track uploaded DBs by db_id -> filesystem path.
+    - Enforce a TTL for uploaded DBs.
+    - Remove stale entries and delete underlying files when expired.
+    """
+
+    def __init__(self, upload_dir: str, ttl_seconds: int) -> None:
+        self.upload_dir = upload_dir
+        self.ttl_seconds = ttl_seconds
+        self._entries: Dict[str, Tuple[str, float]] = {}
+
+        Path(self.upload_dir).mkdir(parents=True, exist_ok=True)
+        log.debug(
+            "Initialized DbUploadStore",
+            extra={
+                "upload_dir": self.upload_dir,
+                "ttl_seconds": self.ttl_seconds,
+            },
+        )
+
+    def _now(self) -> float:
+        return time.time()
+
+    def _is_expired(self, ts: float, now: Optional[float] = None) -> bool:
+        if now is None:
+            now = self._now()
+        return (now - ts) > self.ttl_seconds
+
+    def _gc_locked(self, now: Optional[float] = None) -> None:
+        """
+        Internal garbage collector.
+
+        Removes stale entries and deletes the corresponding files on disk
+        if they still exist.
+        """
+        if now is None:
+            now = self._now()
+
+        to_delete = []
+        for db_id, (path, ts) in list(self._entries.items()):
+            if self._is_expired(ts, now) or (not os.path.exists(path)):
+                to_delete.append((db_id, path))
+
+        for db_id, path in to_delete:
+            self._entries.pop(db_id, None)
+            try:
+                if os.path.exists(path):
+                    os.remove(path)
+                    log.debug(
+                        "Deleted expired uploaded DB file",
+                        extra={"db_id": db_id, "path": path},
+                    )
+            except Exception as exc:
+                # Best-effort cleanup; do not crash the app because of FS issues.
+                log.debug(
+                    "Failed to delete expired uploaded DB file",
+                    extra={"db_id": db_id, "path": path},
+                    exc_info=exc,
+                )
+
+    def cleanup_stale(self) -> None:
+        """
+        Public cleanup entry point.
+
+        Can be called periodically (or on access) to remove expired DBs.
+        """
+        self._gc_locked()
+
+    def register(self, db_id: str, path: str) -> None:
+        """
+        Register a new uploaded DB with the given db_id and filesystem path.
+        """
+        now = self._now()
+        self._entries[db_id] = (path, now)
+        log.debug(
+            "Registered uploaded DB in DbUploadStore",
+            extra={"db_id": db_id, "path": path},
+        )
+        # Optionally clean up old entries as we go.
+        self._gc_locked(now=now)
+
+    def resolve(self, db_id: str) -> Optional[str]:
+        """
+        Resolve db_id to a filesystem path if it exists and is not expired.
+
+        Returns:
+            str path if valid, or None if missing/expired.
+        """
+        self._gc_locked()
+        entry = self._entries.get(db_id)
+        if not entry:
+            return None
+
+        path, ts = entry
+        if self._is_expired(ts):
+            # Expired between last GC and now; treat as missing.
+            self._entries.pop(db_id, None)
+            try:
+                if os.path.exists(path):
+                    os.remove(path)
+            except Exception as exc:
+                log.debug(
+                    "Failed to delete DB file on late-expiration",
+                    extra={"db_id": db_id, "path": path},
+                    exc_info=exc,
+                )
+            return None
+
+        if not os.path.exists(path):
+            # File disappeared; drop the entry.
+            self._entries.pop(db_id, None)
+            return None
+
+        return path
+
+
+# --------------------------------------------------------------------
+# Module-level singleton and legacy helper functions
+# --------------------------------------------------------------------
+
+_DB_UPLOAD_DIR = os.getenv("DB_UPLOAD_DIR", "/tmp/nl2sql_dbs")
+_DB_TTL_SECONDS = int(os.getenv("DB_TTL_SECONDS", "7200"))  # default: 2 hours
+
+_STORE = DbUploadStore(upload_dir=_DB_UPLOAD_DIR, ttl_seconds=_DB_TTL_SECONDS)
 
 
 def register_db(db_id: str, path: str) -> None:
-    """Register new DB in memory (and ensure dir exists)."""
-    _DB_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
-    DB_MAP[db_id] = {"path": path, "ts": time.time()}
-    log.info(f"📦 Registered DB {db_id} -> {path}")
+    """
+    Backwards-compatible helper:
+
+    Register an uploaded DB in the process-wide DbUploadStore.
+    """
+    _STORE.register(db_id, path)
 
 
 def cleanup_stale_dbs() -> None:
-    """Remove expired DBs from /tmp/nl2sql_dbs and memory map."""
-    now = time.time()
-    cutoff = DB_TTL_HOURS * 3600
-    stale_ids = [db_id for db_id, entry in DB_MAP.items() if now - entry["ts"] > cutoff]
-    for db_id in stale_ids:
-        path_str = DB_MAP[db_id]["path"]
-        path = Path(path_str)
-        try:
-            if path.exists():
-                path.unlink()
-                log.info(f"🧹 Deleted stale DB: {path}")
-        except FileNotFoundError:
-            pass
-        DB_MAP.pop(db_id, None)
+    """
+    Backwards-compatible helper:
+
+    Trigger TTL-based cleanup of stale DB entries.
+    """
+    _STORE.cleanup_stale()
 
 
 def get_db_path(db_id: str) -> Optional[str]:
-    """Return full path of an uploaded DB (persistent lookup)."""
-    entry = DB_MAP.get(db_id)
-    if entry:
-        path_str = entry["path"]
-        if Path(path_str).exists():
-            return path_str
-
-    candidates = [
-        _DB_UPLOAD_DIR / f"{db_id}.sqlite",
-        _DB_UPLOAD_DIR / f"{db_id}.db",
-        Path("data/uploads") / f"{db_id}.sqlite",
-        Path("data/uploads") / f"{db_id}.db",
-    ]
-    for p in candidates:
-        if p.exists():
-            log.info(f"🔍 Recovered DB path for {db_id}: {p}")
-            return str(p)
-
-    log.warning(f"⚠️ DB file not found for id={db_id}")
-    return None
+    """
+    Backwards-compatible helper:
+
+    Resolve db_id to a filesystem path if it is still valid.
+    """
+    return _STORE.resolve(db_id)

huggingface.yml

@@ -0,0 +1,3 @@
+sdk: gradio
+app_file: app/app.py
+python_version: "3.10"