""" cBioPortal async helper – returns cohort-level mutation summary. • Works for ANY public molecular-profile ID • One retry on 5xx, returns [] on 404/timeout • Optional CBIO_KEY (JWT) for private studies """ from functools import lru_cache import os, asyncio, httpx _BASE = "https://www.cbioportal.org/api" _URL = _BASE + ( "/molecular-profiles/{profile}/genes/{gene}/mutations?projection=SUMMARY" ) _KEY = os.getenv("CBIO_KEY") # optional @lru_cache(maxsize=256) async def fetch_cbio(gene: str, profile: str = "brca_tcga_pan_can_atlas_2018_mutations" ) -> list[dict]: hdr = {"Accept": "application/json"} if _KEY: hdr["Authorization"] = f"Bearer {_KEY}" async def _once() -> list[dict]: async with httpx.AsyncClient(timeout=10, headers=hdr) as c: r = await c.get(_URL.format(profile=profile, gene=gene)) if r.status_code == 404: # gene not present return [] r.raise_for_status() return r.json() try: return await _once() except httpx.HTTPStatusError: await asyncio.sleep(0.4) # back-off then retry try: return await _once() except Exception: # still bad → empty payload return []