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 threading7import traceback8from concurrent.futures import ThreadPoolExecutor9from datetime import UTC, date, datetime, timedelta10from typing import Literal1112import pyarrow.parquet as pq13from fastapi import APIRouter, Depends, Request14from fastapi.responses import FileResponse15from pydantic import BaseModel, Field1617from findata_api import auth as A18from findata_api.errors import ApiError19from findata_api.ratelimit import record_usage20from findata_core.config import get_settings21from findata_core.db import connection, transaction22from findata_ingest.lake import Lake23from findata_ingest.reader import apply_adjustment, read_bars2425router = APIRouter(prefix="/exports", tags=["exports"])26_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="export")272829class ExportIn(BaseModel):30 dataset_id: str = "market_bars.stock.1m.raw"31 symbols: list[str] = Field(..., min_length=1, max_length=500)32 start: date33 end: date34 adjustment: Literal["raw", "split", "split_div"] = "raw"35 format: Literal["parquet", "csv"] = "parquet"363738def _run(export_id: str, req: dict, principal: str) -> None:39 lake = Lake()40 try:41 with transaction() as conn:42 conn.execute("UPDATE exports SET status='running' WHERE id=%s", (export_id,))43 syms = [s.upper() for s in req["symbols"]]44 table, meta = read_bars(req["dataset_id"], syms, date.fromisoformat(req["start"]), date.fromisoformat(req["end"]), version=req.get("version"))45 if req["adjustment"] != "raw":46 table, _ = apply_adjustment(table, req["adjustment"])47 ext = "parquet" if req["format"] == "parquet" else "csv"48 rel = f"exports/{export_id}.{ext}"49 out = lake.abs(rel)50 if ext == "parquet":51 pq.write_table(table, out, compression="zstd")52 else:53 import pyarrow.csv as pcsv54 pcsv.write_csv(table.drop_columns(["quality_flags"]), out)55 ttl = get_settings().export_ttl_hours56 with transaction() as conn:57 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",58 (rel, out.stat().st_size, table.num_rows, datetime.now(UTC) + timedelta(hours=ttl),59 json.dumps({"dataset_id": req["dataset_id"], "version": meta.get("version"), "files_scanned": meta.get("files_scanned")}), export_id))60 record_usage(principal, rows=table.num_rows, bytes_exported=out.stat().st_size)61 except Exception as e: # noqa: BLE00162 with transaction() as conn:63 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))646566@router.post("", status_code=202)67def create(body: ExportIn, request: Request, p: A.Principal = Depends(A.require("export"))):68 if body.end < body.start:69 raise ApiError(400, "bad_range", "end must be >= start")70 span = (body.end - body.start).days71 if span > 366 * 12:72 raise ApiError(413, "range_too_large", "exports are limited to 12 years per request")73 with connection() as conn:74 d = conn.execute("SELECT id, status, current_version FROM datasets WHERE id=%s", (body.dataset_id,)).fetchone()75 if not d:76 raise ApiError(404, "not_found", "unknown dataset")77 if d["status"] != "active":78 raise ApiError(409, "dataset_not_active", f"dataset status is {d['status']}")79 n = conn.execute("SELECT count(*) FROM exports WHERE principal=%s AND status IN ('queued','running')", (p.label,)).fetchone()["count"]80 if n >= 3:81 raise ApiError(429, "too_many_exports", "at most 3 concurrent exports per principal")82 req = body.model_dump(mode="json")83 req["version"] = d["current_version"]84 with transaction() as conn:85 r = conn.execute("INSERT INTO exports(principal, request, format) VALUES (%s,%s,%s) RETURNING id", (p.label, json.dumps(req), body.format)).fetchone()86 eid = str(r["id"])87 A.audit(request, "export.create", eid, dataset=body.dataset_id, symbols=len(body.symbols), start=str(body.start), end=str(body.end))88 _pool.submit(_run, eid, req, p.label)89 return {"id": eid, "status": "queued", "poll": f"/api/v1/exports/{eid}"}909192@router.get("")93def list_exports(p: A.Principal = Depends(A.require("export"))):94 with connection() as conn:95 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",96 (p.label, p.has("admin"))).fetchall()97 return {"data": rows}9899100@router.get("/{export_id}")101def get(export_id: str, p: A.Principal = Depends(A.require("export"))):102 with connection() as conn:103 r = conn.execute("SELECT * FROM exports WHERE id=%s", (export_id,)).fetchone()104 if not r or (r["principal"] != p.label and not p.has("admin")):105 raise ApiError(404, "not_found", "unknown export")106 r = dict(r)107 if r["status"] == "ready":108 r["download"] = f"/api/v1/exports/{export_id}/download"109 return r110111112@router.get("/{export_id}/download")113def download(export_id: str, request: Request, p: A.Principal = Depends(A.require("export"))):114 with connection() as conn:115 r = conn.execute("SELECT * FROM exports WHERE id=%s", (export_id,)).fetchone()116 if not r or (r["principal"] != p.label and not p.has("admin")):117 raise ApiError(404, "not_found", "unknown export")118 if r["status"] != "ready":119 raise ApiError(409, "not_ready", f"export status is {r['status']}")120 if r["expires_at"] and r["expires_at"] < datetime.now(UTC):121 raise ApiError(410, "expired", "export expired")122 path = Lake().abs(r["path"])123 A.audit(request, "export.download", export_id, bytes=r["bytes"])124 return FileResponse(path, filename=f"findata-{export_id}.{r['format']}", media_type="application/octet-stream")125