1"""Async exports: bounded request validated synchronously, executed in a background thread, Parquet/CSV file served2from lake/exports with expiration. Dataset snapshot (version) is pinned at submission time."""3from __future__ import annotations45import json6import traceback7from concurrent.futures import ThreadPoolExecutor8from datetime import UTC, date, datetime, timedelta9from typing import Literal1011import pyarrow.parquet as pq12from fastapi import APIRouter, Depends, Request13from fastapi.responses import FileResponse14from findata_core.config import get_settings15from findata_core.db import connection, transaction16from findata_ingest.lake import Lake17from findata_ingest.reader import apply_adjustment, read_bars18from pydantic import BaseModel, Field1920from findata_api import auth as A21from findata_api.errors import ApiError22from findata_api.ratelimit import record_usage2324router = APIRouter(prefix="/exports", tags=["exports"])25_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="export")262728class ExportIn(BaseModel):29 dataset_id: str = "market_bars.stock.1m.raw"30 symbols: list[str] = Field(..., min_length=1, max_length=500)31 start: date32 end: date33 adjustment: Literal["raw", "split", "split_div"] = "raw"34 format: Literal["parquet", "csv"] = "parquet"353637def _run(export_id: str, req: dict, principal: str) -> None:38 lake = Lake()39 try:40 with transaction() as conn:41 conn.execute("UPDATE exports SET status='running' WHERE id=%s", (export_id,))42 syms = [s.upper() for s in req["symbols"]]43 table, meta = read_bars(req["dataset_id"], syms, date.fromisoformat(req["start"]), date.fromisoformat(req["end"]), version=req.get("version"))44 if req["adjustment"] != "raw":45 table, _ = apply_adjustment(table, req["adjustment"])46 ext = "parquet" if req["format"] == "parquet" else "csv"47 rel = f"exports/{export_id}.{ext}"48 out = lake.abs(rel)49 if ext == "parquet":50 pq.write_table(table, out, compression="zstd")51 else:52 import pyarrow.csv as pcsv53 pcsv.write_csv(table.drop_columns(["quality_flags"]), out)54 ttl = get_settings().export_ttl_hours55 with transaction() as conn:56 conn.execute("UPDATE exports SET status='ready', path=%s, bytes=%s, row_count=%s, finished_at=now(), expires_at=%s, snapshot=%s WHERE id=%s",57 (rel, out.stat().st_size, table.num_rows, datetime.now(UTC) + timedelta(hours=ttl),58 json.dumps({"dataset_id": req["dataset_id"], "version": meta.get("version"), "files_scanned": meta.get("files_scanned")}), export_id))59 record_usage(principal, rows=table.num_rows, bytes_exported=out.stat().st_size)60 except Exception as e:61 with transaction() as conn:62 conn.execute("UPDATE exports SET status='failed', error=%s, finished_at=now() WHERE id=%s", (f"{e}\n{traceback.format_exc()[-800:]}"[:2000], export_id))636465@router.post("", status_code=202)66def create(body: ExportIn, request: Request, p: A.Principal = Depends(A.require("export"))):67 if body.end < body.start:68 raise ApiError(400, "bad_range", "end must be >= start")69 span = (body.end - body.start).days70 if span > 366 * 12:71 raise ApiError(413, "range_too_large", "exports are limited to 12 years per request")72 with connection() as conn:73 d = conn.execute("SELECT id, status, current_version FROM datasets WHERE id=%s", (body.dataset_id,)).fetchone()74 if not d:75 raise ApiError(404, "not_found", "unknown dataset")76 if d["status"] != "active":77 raise ApiError(409, "dataset_not_active", f"dataset status is {d['status']}")78 n = conn.execute("SELECT count(*) FROM exports WHERE principal=%s AND status IN ('queued','running')", (p.label,)).fetchone()["count"]79 if n >= 3:80 raise ApiError(429, "too_many_exports", "at most 3 concurrent exports per principal")81 req = body.model_dump(mode="json")82 req["version"] = d["current_version"]83 with transaction() as conn:84 r = conn.execute("INSERT INTO exports(principal, request, format) VALUES (%s,%s,%s) RETURNING id", (p.label, json.dumps(req), body.format)).fetchone()85 eid = str(r["id"])86 A.audit(request, "export.create", eid, dataset=body.dataset_id, symbols=len(body.symbols), start=str(body.start), end=str(body.end))87 _pool.submit(_run, eid, req, p.label)88 return {"id": eid, "status": "queued", "poll": f"/api/v1/exports/{eid}"}899091@router.get("")92def list_exports(p: A.Principal = Depends(A.require("export"))):93 with connection() as conn:94 rows = conn.execute("SELECT id, status, request, format, bytes, row_count, created_at, finished_at, expires_at, error, snapshot FROM exports WHERE principal=%s OR %s ORDER BY created_at DESC LIMIT 100",95 (p.label, p.has("admin"))).fetchall()96 return {"data": rows}979899@router.get("/{export_id}")100def get(export_id: str, p: A.Principal = Depends(A.require("export"))):101 with connection() as conn:102 r = conn.execute("SELECT * FROM exports WHERE id=%s", (export_id,)).fetchone()103 if not r or (r["principal"] != p.label and not p.has("admin")):104 raise ApiError(404, "not_found", "unknown export")105 r = dict(r)106 if r["status"] == "ready":107 r["download"] = f"/api/v1/exports/{export_id}/download"108 return r109110111@router.get("/{export_id}/download")112def download(export_id: str, request: Request, p: A.Principal = Depends(A.require("export"))):113 with connection() as conn:114 r = conn.execute("SELECT * FROM exports WHERE id=%s", (export_id,)).fetchone()115 if not r or (r["principal"] != p.label and not p.has("admin")):116 raise ApiError(404, "not_found", "unknown export")117 if r["status"] != "ready":118 raise ApiError(409, "not_ready", f"export status is {r['status']}")119 if r["expires_at"] and r["expires_at"] < datetime.now(UTC):120 raise ApiError(410, "expired", "export expired")121 path = Lake().abs(r["path"])122 A.audit(request, "export.download", export_id, bytes=r["bytes"])123 return FileResponse(path, filename=f"findata-{export_id}.{r['format']}", media_type="application/octet-stream")124