Tony_tech/athena-oracle
3
0
Fork 0
Code Issues 6 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: apply fix/adapter-health-1-2-9 changes (unified retry, failure_class, RSS enabled)

Browse Source 
- adapters/__init__.py: add http_get() unified retry helper + AdapterHTTPError
  + failure_class classification (429/5xx/4xx/error/zero_fetch/ok)
  + last_failure_class on SourceAdapter for pipeline capture
- pipeline.py: ENABLED_SOURCES now includes RSS
  + failure_class rollup in run_log (most severe across all adapters)
  + per-source failure_class in source_stats
- schema.sql: add failure_class column to run_log table

Backport of fix/adapter-health-1-2-9 branch (issues #1, #2, #9).
This commit is contained in:
Epictetus
2026-07-12 04:51:24 +00:00
parent 5cf0b803cb
commit feba2fe7b7
3 changed files with 117 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pipeline.py
+29 -7
View File
@@ -37,7 +37,7 @@ ADAPTERS = {
}
# Default enabled sources
ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface"]
ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface", "rss"]
def init_db(db_path: str, schema_path: str) -> sqlite3.Connection:
@@ -252,9 +252,14 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
entries = adapter.fetch(limit=limit)
except Exception as e:
print(f" ✗ {source_name} failed: {e}")
source_stats[source_name] = {"fetched": 0, "stored": 0, "error": str(e)}
source_stats[source_name] = {"fetched": 0, "stored": 0,
"error": str(e),
"failure_class": "error"}
continue
# Capture classification from the adapter (set by http_get on failure)
fc = getattr(adapter, "last_failure_class", None)
# Add adapter_version to metadata
for entry in entries:
meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"]
@@ -262,7 +267,8 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
entry["raw_metadata"] = json.dumps(meta)
all_entries.extend(entries)
source_stats[source_name] = {"fetched": len(entries), "stored": 0}
source_stats[source_name] = {"fetched": len(entries), "stored": 0,
"failure_class": fc or "ok"}
print(f" Fetched: {len(entries)} entries")
# Small spacing between sources
@@ -294,16 +300,32 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
# Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but
# is still a degraded run — record it so run_log can tell
# "intermittent vs consistently-broken" apart over time.
zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")]
zero = [s for s, st in source_stats.items()
if st.get("fetched", 0) == 0 and not st.get("error")]
notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")]
if zero:
notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}")
notes = "; ".join(notes_parts) or "all sources ok"
# Rollup failure_class (issue #2): most severe across sources.
# Priority: 5xx > 4xx > 429 > error > zero_fetch > ok
rank = {"5xx": 5, "4xx": 4, "429": 3, "error": 2, "zero_fetch": 1, "ok": 0}
classes = [st.get("failure_class", "ok") for st in source_stats.values()]
if any(c in ("5xx", "4xx", "429", "error") for c in classes):
run_fc = max((c for c in classes if c in rank),
key=lambda c: rank[c])
elif zero:
run_fc = "zero_fetch"
else:
run_fc = "ok"
try:
conn.execute("""
INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)
VALUES (?, ?, ?, ?, ?)
""", (len(all_entries), stored, json.dumps(ok), json.dumps(failed), notes))
INSERT INTO run_log (total_fetched, total_stored, sources_ok,
sources_failed, failure_class, notes)
VALUES (?, ?, ?, ?, ?, ?)
""", (len(all_entries), stored, json.dumps(ok), json.dumps(failed),
run_fc, notes))
conn.commit()
except Exception as e:
print(f" ⚠ run_log write failed: {e}")