"""Curation CLI — the ONLY thing that calls the cBioPortal API (ADR-0005, condition C4). Run by a developer, on purpose, rarely:: python -m src.curate paad_tcga python -m src.curate paad_tcga ccle_broad_2019 --dry-run python -m src.curate --list python -m src.curate --check # drift/age report, no re-curation, exit 1 if stale It fetches a study's mutation + discrete-CNV calls plus the clinical attributes the request path needs, classifies everything through the same `variant_status` rules the live path always used, and writes `src/resources/curated/.json`. The Space then answers from that file. **Why a CLI and not a warm cache.** A lazily-populated cache still puts the API on the request path for the first (and post-restart, and post-eviction) caller — it makes the coupling intermittent rather than absent, which is worse to debug and no better under load. Curation is a build step; its output is committed and reviewable, so what the Space serves is exactly what someone looked at. **Politeness (condition C3).** Studies are curated serially with a pause between them, and the client self-throttles. Curation is a handful of requests per study, not a crawl. **Cadence (ADR-0007).** Re-curation is event-driven, not scheduled: `--check` costs one metadata request per study and tells you whether upstream re-imported the study since we curated it. Only then is a re-pull worth making. `--check` never writes — refreshing an artifact stays a reviewed commit, because that reviewability is the whole reason the artifacts are in git. """ from __future__ import annotations import argparse import sys import time from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from licenses import is_open_access, study_terms # noqa: E402 from src.workflows import cbioportal_io, curated_store # noqa: E402 from src.workflows.subtype_join import SUBTYPE_ATTRIBUTES # noqa: E402 from src.workflows.variant_status import curate_from_cbioportal # noqa: E402 # Everything the request path may need to resolve offline: lineage + every subtype label. CLINICAL_ATTRIBUTES = curated_store.LINEAGE_ATTRIBUTES + SUBTYPE_ATTRIBUTES # Pause between studies — polite, and irrelevant to the request path since this is offline work. INTER_STUDY_PAUSE_S = 2.0 def fetch_clinical(study_id: str) -> dict[str, dict[str, str]]: """Pull the clinical attributes the artifact must carry. An attribute a study does not have simply comes back empty and is omitted — that absence is meaningful downstream (`find_subtype_labels` reads it as "no label in this cohort"). """ out: dict[str, dict[str, str]] = {} for attr in CLINICAL_ATTRIBUTES: try: rows = cbioportal_io.clinical_data(study_id, [attr]) except Exception as exc: # noqa: BLE001 — one missing attribute must not sink the run print(f" ! {attr}: fetch failed ({type(exc).__name__}: {exc}) — skipped") continue values = {r["sampleId"]: r["value"] for r in rows if (r.get("value") or "").strip()} if values: out[attr] = values print(f" · {attr}: {len(values)} samples") return out class ControlledAccessError(RuntimeError): """A study is not open-access, so we refuse to curate it (condition C2). Refusing at curation time is the only gate that matters: an artifact that should never have existed is one nobody re-examines. Open-access somatic *status* is the whole scope — controlled tiers (germline, raw reads) are not something to handle carefully, they are something not to fetch. """ def curate(study_id: str, *, dry_run: bool = False) -> dict: """Curate one study end-to-end and (unless `dry_run`) write its artifact. Refuses non-open-access studies before fetching any variant data (C2). """ print(f"[curate] {study_id}") metadata = cbioportal_io.study_metadata(study_id) terms = study_terms(study_id, metadata) if not is_open_access(terms): raise ControlledAccessError( f"{study_id}: access_tier={terms['access_tier']} " f"(publicStudy={terms['public_study']}, readPermission={terms['read_permission']}). " "Only open-access studies are curated (ADR-0005 condition C2)." ) if metadata.get("citation"): terms["source_citation"] = metadata["citation"] if metadata.get("pmid"): terms["pmid"] = metadata["pmid"] print(f" · access={terms['access_tier']}, license=ODbL (per-study terms carried)") matrix = curate_from_cbioportal(study_id) print( f" · {len(matrix.samples)} samples × {len(matrix.genes)} genes; " f"modalities={ {k: v for k, v in matrix.modalities.items()} }" ) clinical = fetch_clinical(study_id) payload = curated_store.build_payload( matrix, study_id=study_id, clinical=clinical, terms=terms, import_date=metadata.get("importDate"), ) counts = payload["counts"] print( f" · {counts['n_mutated_cells']} mutated cells, {counts['n_cnv_cells']} CNV cells, " f"{counts.get('n_sv_cells', 0)} SV cells" + (f" ({payload.get('sv_annotation')})" if payload.get("sv_annotation") else "") ) if dry_run: print(" · dry run — not written") return payload path = curated_store.write(payload) # relative when it is under the repo (the normal case), absolute otherwise (e.g. tests # curating into a temp dir) — a display detail must never sink a curation run. try: shown = path.relative_to(PROJECT_ROOT) except ValueError: shown = path print(f" ✓ {shown}") return payload def check(study_ids: list[str] | None = None) -> tuple[list[dict], int]: """Staleness CHECK — report drift without re-curating anything (ADR-0007). One `/studies/` request per study: compare upstream's `importDate` against the one recorded when we curated, and report the artifact's age. That is the whole point of driving off an upstream signal rather than a calendar — a cohort that upstream has not re-imported needs no refresh no matter how old our file is, and re-pulling it on a monthly timer would spend the C3 politeness budget to rewrite a byte-identical artifact. This is a maintainer/CI tool, not a request-path one: it *reads* artifacts and never writes. Re-curation stays a deliberate, reviewed commit — a checker that repaired what it found would put the API back on an automatic path, which is exactly what C4 removed. Returns `(rows, exit_code)`. Exit code 1 on any drift or `stale` artifact, so CI can fail. """ studies = list(study_ids) if study_ids else curated_store.list_curated() if not studies: print("(no curated studies)") return [], 0 rows: list[dict] = [] for i, study in enumerate(studies): if i: time.sleep(INTER_STUDY_PAUSE_S) fresh = curated_store.freshness(study) if fresh is None: rows.append({"study": study, "state": "not_curated"}) continue row = { "study": study, "age_days": fresh["age_days"], "level": fresh["level"], "curated_import_date": fresh["source_import_date"], } try: row["upstream_import_date"] = cbioportal_io.study_metadata(study).get("importDate") except Exception as exc: # noqa: BLE001 — an unreachable API is a check failure, not drift row["state"] = "check_failed" row["error"] = f"{type(exc).__name__}: {exc}" rows.append(row) continue if not fresh["drift_comparable"]: # Pre-ADR-0007 artifact: no recorded importDate to compare against. Say so — an # unknown is not a "no drift". row["state"] = "drift_unknown" elif row["upstream_import_date"] != row["curated_import_date"]: row["state"] = "drift" else: row["state"] = "up_to_date" rows.append(row) print(f"{'study':<24} {'state':<14} {'age':>6} upstream importDate") for row in rows: age = "?" if row.get("age_days") is None else f"{row['age_days']}d" print( f"{row['study']:<24} {row.get('state', '?'):<14} {age:>6} " f"{row.get('upstream_import_date') or '-'}" + (f" (curated from {row['curated_import_date']})" if row.get("state") == "drift" else "") + (f" {row.get('error')}" if row.get("error") else "") ) drifted = [r["study"] for r in rows if r.get("state") == "drift"] stale = [r["study"] for r in rows if r.get("level") == "stale"] unknown = [r["study"] for r in rows if r.get("state") == "drift_unknown"] if unknown: print( f"\nNo recorded upstream importDate for: {', '.join(unknown)} — curated before " "ADR-0007. Drift cannot be established for these until they are next re-curated." ) if drifted: print(f"\nUpstream has re-imported since we curated: {', '.join(drifted)}") print(f"Review and re-curate deliberately: python -m src.curate {' '.join(drifted)}") if stale: print( f"\nPast the {curated_store.STALE_AFTER_DAYS}-day horizon: {', '.join(stale)} " "(answers still serve, and say their age)." ) if not drifted and not stale: print("\nAll curated artifacts are within the review horizon and match upstream.") return rows, (1 if (drifted or stale) else 0) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="python -m src.curate", description=__doc__.splitlines()[0] ) parser.add_argument("studies", nargs="*", help="cBioPortal study ids, e.g. paad_tcga") parser.add_argument("--list", action="store_true", help="list already-curated studies") parser.add_argument("--dry-run", action="store_true", help="fetch and report, do not write") parser.add_argument( "--check", action="store_true", help="report artifact age + upstream drift without re-curating; exit 1 if any drifted " "or is past the stale horizon (CI-friendly)", ) args = parser.parse_args(argv) if args.list: curated = curated_store.list_curated() print("\n".join(curated) if curated else "(no curated studies)") return 0 if args.check: print(f"[check] User-Agent: {cbioportal_io.user_agent()}\n") return check(args.studies)[1] if not args.studies: parser.error("give at least one study id, --check, or --list") # C3: show what we are identifying ourselves as, so an operator can see it in one place. # The +URL in the User-Agent is the contact channel; no extra config is required. print(f"[curate] User-Agent: {cbioportal_io.user_agent()}\n") failures = [] for i, study in enumerate(args.studies): if i: time.sleep(INTER_STUDY_PAUSE_S) try: curate(study, dry_run=args.dry_run) except Exception as exc: # noqa: BLE001 print(f" ✗ {study}: {type(exc).__name__}: {exc}") failures.append(study) if failures: print(f"\nFailed: {', '.join(failures)}") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())