Update mcp/targets.py
Browse files- mcp/targets.py +33 -10
mcp/targets.py
CHANGED
@@ -1,14 +1,19 @@
|
|
1 |
-
|
2 |
-
|
3 |
-
|
|
|
|
|
|
|
4 |
"""
|
|
|
5 |
|
6 |
-
import
|
|
|
7 |
from typing import List, Dict
|
8 |
|
9 |
_OT_URL = "https://api.platform.opentargets.org/api/v4/graphql"
|
10 |
|
11 |
-
|
12 |
"""
|
13 |
query Assoc($gene: String!, $size: Int!) {
|
14 |
associations(geneSymbol: $gene, size: $size) {
|
@@ -24,9 +29,27 @@ _QUERY_TEMPLATE = textwrap.dedent(
|
|
24 |
"""
|
25 |
)
|
26 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
27 |
async def fetch_ot_associations(gene_symbol: str, *, size: int = 30) -> List[Dict]:
|
28 |
-
|
29 |
-
|
30 |
-
|
31 |
-
|
32 |
-
|
|
|
|
|
|
1 |
+
#!/usr/bin/env python3
|
2 |
+
"""MedGenesis – **Open Targets** GraphQL helper (async, cached).
|
3 |
+
Updated version adds:
|
4 |
+
* Retry with exponential back‑off (2→4→8 s)
|
5 |
+
* 12‑h LRU cache (256 genes)
|
6 |
+
* Graceful empty‑list fallback & doc‑string examples.
|
7 |
"""
|
8 |
+
from __future__ import annotations
|
9 |
|
10 |
+
import asyncio, textwrap, httpx
|
11 |
+
from functools import lru_cache
|
12 |
from typing import List, Dict
|
13 |
|
14 |
_OT_URL = "https://api.platform.opentargets.org/api/v4/graphql"
|
15 |
|
16 |
+
_QUERY = textwrap.dedent(
|
17 |
"""
|
18 |
query Assoc($gene: String!, $size: Int!) {
|
19 |
associations(geneSymbol: $gene, size: $size) {
|
|
|
29 |
"""
|
30 |
)
|
31 |
|
32 |
+
async def _post(payload: Dict, retries: int = 3) -> Dict:
|
33 |
+
"""POST with back‑off retry (2×, 4×); returns JSON."""
|
34 |
+
delay = 2
|
35 |
+
last_resp = None
|
36 |
+
for _ in range(retries):
|
37 |
+
async with httpx.AsyncClient(timeout=15) as cli:
|
38 |
+
last_resp = await cli.post(_OT_URL, json=payload)
|
39 |
+
if last_resp.status_code == 200:
|
40 |
+
return last_resp.json()
|
41 |
+
await asyncio.sleep(delay)
|
42 |
+
delay *= 2
|
43 |
+
# Raise from final attempt if still failing
|
44 |
+
last_resp.raise_for_status() # type: ignore
|
45 |
+
|
46 |
+
|
47 |
+
@lru_cache(maxsize=256)
|
48 |
async def fetch_ot_associations(gene_symbol: str, *, size: int = 30) -> List[Dict]:
|
49 |
+
"""Return association rows for *gene_symbol* (or []). Example::
|
50 |
+
|
51 |
+
rows = await fetch_ot_associations("TP53", size=20)
|
52 |
+
"""
|
53 |
+
payload = {"query": _QUERY, "variables": {"gene": gene_symbol, "size": size}}
|
54 |
+
data = await _post(payload)
|
55 |
+
return data.get("data", {}).get("associations", {}).get("rows", [])
|