|
|
|
|
|
"""MedGenesis – **Open Targets** GraphQL helper (async, cached). |
|
|
Updated version adds: |
|
|
* Retry with exponential back‑off (2→4→8 s) |
|
|
* 12‑h LRU cache (256 genes) |
|
|
* Graceful empty‑list fallback & doc‑string examples. |
|
|
""" |
|
|
from __future__ import annotations |
|
|
|
|
|
import asyncio, textwrap, httpx |
|
|
from functools import lru_cache |
|
|
from typing import List, Dict |
|
|
|
|
|
_OT_URL = "https://api.platform.opentargets.org/api/v4/graphql" |
|
|
|
|
|
_QUERY = textwrap.dedent( |
|
|
""" |
|
|
query Assoc($gene: String!, $size: Int!) { |
|
|
associations(geneSymbol: $gene, size: $size) { |
|
|
rows { |
|
|
score |
|
|
datatypeId |
|
|
datasourceId |
|
|
disease { id name } |
|
|
target { id symbol } |
|
|
} |
|
|
} |
|
|
} |
|
|
""" |
|
|
) |
|
|
|
|
|
async def _post(payload: Dict, retries: int = 3) -> Dict: |
|
|
"""POST with back‑off retry (2×, 4×); returns JSON.""" |
|
|
delay = 2 |
|
|
last_resp = None |
|
|
for _ in range(retries): |
|
|
async with httpx.AsyncClient(timeout=15) as cli: |
|
|
last_resp = await cli.post(_OT_URL, json=payload) |
|
|
if last_resp.status_code == 200: |
|
|
return last_resp.json() |
|
|
await asyncio.sleep(delay) |
|
|
delay *= 2 |
|
|
|
|
|
last_resp.raise_for_status() |
|
|
|
|
|
|
|
|
@lru_cache(maxsize=256) |
|
|
async def fetch_ot_associations(gene_symbol: str, *, size: int = 30) -> List[Dict]: |
|
|
"""Return association rows for *gene_symbol* (or []). Example:: |
|
|
|
|
|
rows = await fetch_ot_associations("TP53", size=20) |
|
|
""" |
|
|
payload = {"query": _QUERY, "variables": {"gene": gene_symbol, "size": size}} |
|
|
data = await _post(payload) |
|
|
return data.get("data", {}).get("associations", {}).get("rows", []) |
|
|
|
