fundamentals: remember 404 listings in the checkpoint so resumes do not retry them
1 changed file +3 −1
modified
services/ingest/src/findata_ingest/tasks/fundamentals.py
+3 −1
@@ -32,6 +32,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 32 | stats = {"tickers": 0, "facts_new": 0, "facts_unchanged": 0, "facts_restated": 0, "earnings": 0, "errors": 0, "pit_fallback": 0} | 32 | stats = {"tickers": 0, "facts_new": 0, "facts_unchanged": 0, "facts_restated": 0, "earnings": 0, "errors": 0, "pit_fallback": 0} | ||
| 33 | ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {} | 33 | ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {} | ||
| 34 | done = set(ck.get("done") or []) | 34 | done = set(ck.get("done") or []) | ||
| 35 | + no_data = set(ck.get("no_data") or []) | ||||
| 35 | for tk in tickers: | 36 | for tk in tickers: | ||
| 36 | if tk in done: | 37 | if tk in done: | ||
| 37 | continue | 38 | continue | ||
@@ -78,8 +79,9 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 78 | stats["earnings"] += len(earnings) | 79 | stats["earnings"] += len(earnings) | ||
| 79 | stats["tickers"] += 1 | 80 | stats["tickers"] += 1 | ||
| 80 | done.add(tk) | 81 | done.add(tk) | ||
| 81 | − queue.set_checkpoint(f"job:{job['id']}", "state", {"done": sorted(done)}) | 82 | + queue.set_checkpoint(f"job:{job['id']}", "state", {"done": sorted(done), "no_data": sorted(no_data)}) | ||
| 82 | queue.progress(job["id"], {**stats, "last": tk}) | 83 | queue.progress(job["id"], {**stats, "last": tk}) | ||
| 83 | fr.close() | 84 | fr.close() | ||
| 85 | + stats["no_statements_404"] = len(no_data) | ||||
| 84 | log.info("fundamentals.done", **stats) | 86 | log.info("fundamentals.done", **stats) | ||
| 85 | return stats | 87 | return stats | ||
| 86 | 88 | ||||