File size: 1,527 Bytes
64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 64eb751 20e7231 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | """
cBioPortal helper · REST v4 · zero-crash
* Public instance needs **no** API-key :contentReference[oaicite:0]{index=0}
* Optional Bearer token (`CBIO_KEY`) for private portals (e.g. GENIE) :contentReference[oaicite:1]{index=1}
* One retry on 5xx / ReadTimeout, else returns []
"""
from __future__ import annotations
import os, asyncio, httpx
from functools import lru_cache
from typing import List, Dict
_TOKEN = os.getenv("CBIO_KEY") # optional
_HDR = {"Accept": "application/json"}
if _TOKEN:
_HDR["Authorization"] = f"Bearer {_TOKEN}"
_URL = ("https://www.cbioportal.org/api"
"/molecular-profiles/{profile}/genes/{gene}/mutations?projection=SUMMARY"
) # swagger ref :contentReference[oaicite:2]{index=2}
@lru_cache(maxsize=256)
async def fetch_cbio(
gene : str,
profile : str = "brca_tcga_pan_can_atlas_2018_mutations"
) -> List[Dict]:
async def _hit() -> 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:
return [] # gene absent
r.raise_for_status()
return r.json()
try:
return await _hit()
except (httpx.HTTPStatusError, httpx.ReadTimeout):
await asyncio.sleep(0.5) # one back-off
try:
return await _hit()
except Exception:
return [] # graceful fallback
|