cli: worker --role, plan-all, backfill-all, options, fundamentals --all-active
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 changed file +51 −3
modified
services/ingest/src/findata_ingest/cli.py
+51 −3
@@ -44,10 +44,58 @@ def discover(source: str = "all") -> None: | |||||
| 44 | 44 | ||||
| 45 | 45 | ||||
| 46 | @app.command() | 46 | @app.command() | ||
| 47 | −def worker(once: bool = False, no_scheduler: bool = False, kinds: str | None = None) -> None: | 47 | +def worker(once: bool = False, no_scheduler: bool = False, kinds: str | None = None, role: str = "primary") -> None: | ||
| 48 | − """Run the single ingest worker (+ scheduler).""" | 48 | + """Run an ingest worker: role=primary (lake lock + scheduler) or role=bulk (parallel backfill/options/fundamentals).""" | ||
| 49 | from findata_ingest.worker import main | 49 | from findata_ingest.worker import main | ||
| 50 | − main(once=once, kinds=kinds.split(",") if kinds else None, with_scheduler=not no_scheduler) | 50 | + main(once=once, kinds=kinds.split(",") if kinds else None, with_scheduler=not no_scheduler, role=role) | ||
| 51 | + | ||||
| 52 | + | ||||
| 53 | +@ingest_app.command("plan-all") | ||||
| 54 | +def ingest_plan_all(types: str | None = None, no_options: bool = False, no_delisted: bool = False) -> None: | ||||
| 55 | + """Show the complete vendor download plan (counts per dataset), without enqueuing anything.""" | ||||
| 56 | + from collections import Counter | ||||
| 57 | + | ||||
| 58 | + from findata_market_firstrate import build_full_plan | ||||
| 59 | + plan = build_full_plan(types.split(",") if types else None, include_options=not no_options, include_delisted=not no_delisted) | ||||
| 60 | + typer.echo(f"{len(plan)} jobs") | ||||
| 61 | + for ds, n in sorted(Counter(j["dataset_id"] for j in plan).items()): | ||||
| 62 | + typer.echo(f" {ds:<48} {n}") | ||||
| 63 | + | ||||
| 64 | + | ||||
| 65 | +@ingest_app.command("backfill-all") | ||||
| 66 | +def ingest_backfill_all(types: str | None = None, no_options: bool = False, no_delisted: bool = False, yes: bool = False) -> None: | ||||
| 67 | + """Enqueue the complete vendor download (every licensed archive). Needs the cold lake root (external SSD).""" | ||||
| 68 | + from findata_core.config import get_settings | ||||
| 69 | + from findata_market_firstrate import build_full_plan | ||||
| 70 | + | ||||
| 71 | + from findata_ingest import queue | ||||
| 72 | + from findata_ingest.lake import Lake | ||||
| 73 | + lake = Lake() | ||||
| 74 | + s = get_settings() | ||||
| 75 | + free_hot = lake.free_gb() | ||||
| 76 | + free_cold = lake.free_gb(lake.cold) if lake.cold else None | ||||
| 77 | + typer.echo(f"hot root {lake.root}: {free_hot:.0f} GB free · cold root {lake.cold}: {free_cold if free_cold is None else round(free_cold)} GB free") | ||||
| 78 | + if not lake.cold: | ||||
| 79 | + typer.echo("refusing: FINDATA_LAKE_COLD is not configured (the full catalog needs the external SSD)") | ||||
| 80 | + raise typer.Exit(2) | ||||
| 81 | + if free_cold < 1200 or free_hot < 300: | ||||
| 82 | + typer.echo("refusing: need >= 1200 GB free on the cold root and >= 300 GB on the hot root for the full catalog") | ||||
| 83 | + raise typer.Exit(2) | ||||
| 84 | + plan = build_full_plan(types.split(",") if types else None, include_options=not no_options, include_delisted=not no_delisted) | ||||
| 85 | + if not yes and not typer.confirm(f"enqueue {len(plan)} jobs?"): | ||||
| 86 | + raise typer.Exit(1) | ||||
| 87 | + n = 0 | ||||
| 88 | + for j in plan: | ||||
| 89 | + if queue.enqueue(j["kind"], j["dataset_id"], j["params"], j["priority"], dedupe_key=j["dedupe_key"], created_by="cli:backfill-all", max_attempts=4): | ||||
| 90 | + n += 1 | ||||
| 91 | + typer.echo(f"enqueued {n} new jobs ({len(plan) - n} already queued/running); bulk workers configured: {s.bulk_workers}") | ||||
| 92 | + | ||||
| 93 | + | ||||
| 94 | +@ingest_app.command("options") | ||||
| 95 | +def ingest_options(year: int | None = None, quarter: str | None = None, month: bool = False) -> None: | ||||
| 96 | + from findata_ingest import queue | ||||
| 97 | + params = {"period": "month"} if month else {"year": year, "quarter": quarter} | ||||
| 98 | + typer.echo(queue.enqueue("options", "options.eod", params, priority=55, created_by="cli")) | ||||
| 51 | 99 | ||||
| 52 | 100 | ||||
| 53 | @ingest_app.command("metadata") | 101 | @ingest_app.command("metadata") | ||
| 54 | 102 | ||||