1from __future__ import annotations23import time45from fastapi import APIRouter, Depends, Request6from findata_core.config import get_settings7from findata_ingest.reader import dataset_views, run_sql8from pydantic import BaseModel, Field910from findata_api import auth as A11from findata_api import metrics as M12from findata_api.errors import ApiError13from findata_api.ratelimit import record_usage1415router = APIRouter(prefix="/query", tags=["query"])161718def _used(sql: str, views: dict) -> list[str]:19 from findata_ingest.reader import _referenced_views20 return _referenced_views(sql, views)212223class QueryIn(BaseModel):24 sql: str = Field(..., max_length=20000)25 max_rows: int = Field(1000, ge=1, le=100_000)262728@router.post("")29def query(body: QueryIn, request: Request, p: A.Principal = Depends(A.require("query"))):30 s = get_settings()31 views = dataset_views()32 t0 = time.monotonic()33 try:34 tbl, meta = run_sql(body.sql, min(body.max_rows, s.query_max_rows), s.query_timeout_seconds, views)35 except ValueError as e:36 raise ApiError(400, "query_rejected", str(e)) from None37 except TimeoutError as e:38 raise ApiError(408, "query_timeout", str(e)) from None39 except Exception as e:40 raise ApiError(400, "query_error", str(e)[:500]) from None41 rows = tbl.to_pylist()42 for r in rows:43 for k, v in r.items():44 if hasattr(v, "isoformat"):45 r[k] = v.isoformat()46 M.ROWS.labels("query").inc(len(rows))47 record_usage(p.label, rows=len(rows))48 A.audit(request, "query.sql", None, rows=len(rows), ms=int((time.monotonic() - t0) * 1000), sql=body.sql[:500])49 return {"columns": [{"name": f.name, "type": str(f.type)} for f in tbl.schema], "rows": rows,50 "meta": {**meta, "rows": len(rows), "elapsed_ms": int((time.monotonic() - t0) * 1000),51 "views_used": meta.get("views", _used(body.sql, views)), "engine": "duckdb (read-only, sandboxed, 1 statement; symbol-scoped statements resolve files from the catalog, others scan the directory)"}}525354@router.get("/views")55def views(p: A.Principal = Depends(A.require("query"))):56 v = dataset_views()57 return {"views": [{"name": k, "kind": spec.get("kind"), "layouts": spec.get("layouts"), "root": spec.get("root")} for k, spec in sorted(v.items())],58 "note": "Fast path: statements with symbol='X' / symbol IN (...) / underlying='X' (plus optional year=, session_date/trade_date ranges) "59 "get their file list from the catalog in milliseconds. Without a symbol predicate the view is a Hive directory scan of the whole "60 "dataset (30–60 s to list hundreds of thousands of files; may time out).",61 "examples": ["SELECT session_date, min(low) lo, max(high) hi, sum(volume) vol FROM market_bars_stock_1m_raw WHERE symbol='AAPL' AND year=2024 GROUP BY 1 ORDER BY 1",62 "SELECT trade_date, strike, call_put, bid, ask, open_interest, delta FROM options_eod WHERE underlying='SPY' AND period='2024q1' AND expiry=DATE '2024-03-28' ORDER BY trade_date, strike LIMIT 100",63 "SELECT symbol, count(*) bars FROM market_bars_futures_1m_continuous_ratio WHERE symbol IN ('ES','NQ') GROUP BY 1"]}64