1from __future__ import annotations23import json4from datetime import datetime5from typing import Any67from fastapi import APIRouter, Depends, Query, Request8from findata_core.config import get_settings9from findata_core.db import connection, transaction10from findata_ingest import queue11from findata_ingest.lake import Lake12from pydantic import BaseModel, Field1314from findata_api import auth as A15from findata_api.errors import ApiError1617router = APIRouter(prefix="/admin", tags=["admin"])181920class JobIn(BaseModel):21 kind: str22 dataset_id: str | None = None23 params: dict[str, Any] = Field(default_factory=dict)24 priority: int = 50252627class KeyIn(BaseModel):28 name: str = Field(..., min_length=1, max_length=60)29 scopes: list[str]30 rate_limit_per_minute: int = Field(600, ge=1, le=10000)31 expires_at: datetime | None = None323334class TrackedIn(BaseModel):35 tickers: list[str]363738@router.get("/jobs")39def jobs(status: str | None = None, kind: str | None = None, limit: int = Query(100, le=1000), p: A.Principal = Depends(A.require("admin"))):40 with connection() as conn:41 rows = conn.execute("SELECT id, kind, dataset_id, params, priority, status, attempts, max_attempts, next_run_at, created_at, started_at, finished_at, last_error, progress, result, created_by "42 "FROM ingest_jobs WHERE (%s::text IS NULL OR status=%s) AND (%s::text IS NULL OR kind=%s) ORDER BY created_at DESC LIMIT %s",43 (status, status, kind, kind, limit)).fetchall()44 return {"data": rows}454647@router.post("/jobs", status_code=201)48def enqueue(body: JobIn, request: Request, p: A.Principal = Depends(A.require("admin"))):49 from findata_ingest.tasks import REGISTRY50 if body.kind not in REGISTRY:51 raise ApiError(400, "bad_kind", f"kind must be one of {sorted(REGISTRY)}")52 if body.kind == "backfill":53 lake = Lake()54 if lake.free_gb() < get_settings().lake_hard_min_free_gb * 1.5:55 raise ApiError(409, "disk_budget", "not enough free space to start a backfill")56 jid = queue.enqueue(body.kind, body.dataset_id, body.params, body.priority, created_by=p.email or p.label)57 A.audit(request, "admin.enqueue", jid, kind=body.kind, dataset=body.dataset_id, params=body.params)58 return {"id": jid}596061@router.post("/jobs/{job_id}/retry")62def retry(job_id: str, request: Request, p: A.Principal = Depends(A.require("admin"))):63 with transaction() as conn:64 r = conn.execute("UPDATE ingest_jobs SET status='queued', attempts=0, next_run_at=now(), last_error=NULL WHERE id=%s AND status IN ('failed','dead','cancelled') RETURNING id", (job_id,)).fetchone()65 if not r:66 raise ApiError(409, "not_retryable", "job is not in a retryable state")67 A.audit(request, "admin.retry", job_id)68 return {"ok": True}697071@router.post("/jobs/{job_id}/cancel")72def cancel(job_id: str, request: Request, p: A.Principal = Depends(A.require("admin"))):73 with transaction() as conn:74 r = conn.execute("UPDATE ingest_jobs SET status='cancelled', finished_at=now() WHERE id=%s AND status='queued' RETURNING id", (job_id,)).fetchone()75 if not r:76 raise ApiError(409, "not_cancellable", "only queued jobs can be cancelled")77 A.audit(request, "admin.cancel", job_id)78 return {"ok": True}798081@router.get("/keys")82def keys(p: A.Principal = Depends(A.require("admin"))):83 with connection() as conn:84 rows = conn.execute("SELECT k.id, k.name, k.prefix, k.scopes, k.rate_limit_per_minute, k.expires_at, k.last_used_at, k.revoked_at, k.created_at, u.email "85 "FROM api_keys k JOIN users u ON u.id=k.user_id ORDER BY k.created_at DESC").fetchall()86 usage = conn.execute("SELECT principal, day, requests, rows_returned, bytes_exported FROM usage_daily WHERE day > current_date - 14 ORDER BY day DESC").fetchall()87 return {"data": rows, "usage": usage}888990@router.post("/keys", status_code=201)91def create_key(body: KeyIn, request: Request, p: A.Principal = Depends(A.require("admin"))):92 try:93 out = A.new_api_key(p.id, body.name, body.scopes, body.rate_limit_per_minute, body.expires_at)94 except ValueError as e:95 raise ApiError(400, "bad_scopes", str(e)) from None96 A.audit(request, "admin.key_create", out["id"], name=body.name, scopes=body.scopes)97 return out9899100@router.delete("/keys/{key_id}")101def revoke_key(key_id: str, request: Request, p: A.Principal = Depends(A.require("admin"))):102 with transaction() as conn:103 conn.execute("UPDATE api_keys SET revoked_at=now() WHERE id=%s", (key_id,))104 A.audit(request, "admin.key_revoke", key_id)105 return {"ok": True}106107108@router.get("/audit")109def audit_log(limit: int = Query(200, le=2000), action: str | None = None, p: A.Principal = Depends(A.require("admin"))):110 with connection() as conn:111 rows = conn.execute("SELECT * FROM audit_log WHERE (%s::text IS NULL OR action LIKE %s) ORDER BY at DESC LIMIT %s", (action, (action or "") + "%", limit)).fetchall()112 return {"data": rows}113114115@router.get("/storage")116def storage(p: A.Principal = Depends(A.require("admin"))):117 lake = Lake()118 s = get_settings()119 with connection() as conn:120 per_ds = conn.execute("SELECT dataset_id, layout, count(*) files, sum(bytes) bytes, sum(row_count) rows FROM lake_files WHERE version_removed IS NULL GROUP BY 1,2 ORDER BY 1,2").fetchall()121 raw = conn.execute("SELECT status, count(*) n, sum(bytes) bytes FROM raw_archives GROUP BY status").fetchall()122 exports = conn.execute("SELECT status, count(*) n, coalesce(sum(bytes),0) bytes FROM exports GROUP BY status").fetchall()123 retired = conn.execute("SELECT count(*) n, coalesce(sum(bytes),0) bytes FROM lake_files WHERE version_removed IS NOT NULL").fetchone()124 db = conn.execute("SELECT pg_database_size(current_database()) bytes").fetchone()125 backup = conn.execute("SELECT value, updated_at FROM settings_kv WHERE key='last_backup'").fetchone()126 from findata_api.health import lake_usage_bytes127 usage = lake_usage_bytes(conn)128 import os129 st = os.statvfs(lake.root)130 return {"lake_root": str(lake.root), "volume": {"total_gb": round(st.f_blocks * st.f_frsize / 1e9, 1), "free_gb": round(st.f_bavail * st.f_frsize / 1e9, 1)},131 "budget": {"soft_limit_gb": s.lake_soft_limit_gb, "hard_min_free_gb": s.lake_hard_min_free_gb},132 "curated_bytes": usage["curated"], "raw_bytes": usage["raw"], "retired_bytes": usage["retired"],133 "exports_bytes": sum(int(e["bytes"] or 0) for e in exports), "per_dataset": per_ds,134 "cold_root": str(lake.cold) if lake.cold else None, "cold_free_gb": round(lake.free_gb(lake.cold), 1) if lake.cold else None, "raw_archives": raw, "exports": exports, "retired_files": retired,135 "postgres_bytes": db["bytes"], "last_backup": backup,136 "capacity_model": {"bytes_per_1m_bar_curated": 13, "note": "see docs/DISCOVERY.md §5 — measured on the prior lake; recomputed from per_dataset above"}}137138139@router.get("/settings")140def get_settings_(p: A.Principal = Depends(A.require("admin"))):141 with connection() as conn:142 rows = conn.execute("SELECT key, value, updated_at FROM settings_kv WHERE key NOT LIKE 'secret%'").fetchall()143 s = get_settings()144 return {"settings": rows, "env": {"env": s.env, "lake": str(s.lake), "public_url": s.public_url, "backup_target": s.backup_target,145 "vendor_configured": bool(s.frd_userid.get_secret_value()), "sec_edgar_configured": bool(s.sec_edgar_user_agent),146 "ai_providers": {"anthropic": bool(s.anthropic_api_key.get_secret_value()), "openai": bool(s.openai_api_key.get_secret_value())}}}147148149@router.put("/settings/tracked-tickers")150def tracked(body: TrackedIn, request: Request, p: A.Principal = Depends(A.require("admin"))):151 tk = sorted({t.upper().strip() for t in body.tickers if t.strip()})[:2000]152 with transaction() as conn:153 conn.execute("INSERT INTO settings_kv(key, value) VALUES ('fundamentals_tracked_tickers', %s) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()", (json.dumps(tk),))154 A.audit(request, "admin.tracked_tickers", None, count=len(tk))155 return {"tickers": tk}156157158@router.post("/discover")159def discover(request: Request, p: A.Principal = Depends(A.require("admin"))):160 """Sanitized live vendor probe (no bulk download)."""161 from findata_fundamentals import SecEdgarConnector162 from findata_market_firstrate import FirstRateMarketConnector163 out = {"firstrate": FirstRateMarketConnector().discover(), "sec_edgar": SecEdgarConnector().status()}164 A.audit(request, "admin.discover")165 return out166