File size: 1,732 Bytes
b6b0f60 93e7f22 b6b0f60 93e7f22 b6b0f60 93e7f22 b6b0f60 93e7f22 b6b0f60 93e7f22 b6b0f60 |
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 45 46 47 48 49 50 51 52 53 54 55 56 |
#!/usr/bin/env python3
"""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
# Raise from final attempt if still failing
last_resp.raise_for_status() # type: ignore
@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", [])
|