File size: 2,325 Bytes
3333933 758d4b9 3333933 758d4b9 3333933 758d4b9 3333933 758d4b9 3333933 758d4b9 3333933 |
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 57 58 59 60 61 62 63 64 65 |
#!/usr/bin/env python3
"""MedGenesis – Human Phenotype Ontology (HPO) async helper.
Adds reliability & caching over the minimal original version.
Capabilities
~~~~~~~~~~~~
* `get_hpo(term_id)` – fetch full term JSON (`HP:XXXXXXX`), cached 24 h.
* `search_hpo(term)` – simple keyword search, returns top hit (id + name).
API docs: https://hpo.jax.org/api
"""
from __future__ import annotations
import httpx, asyncio
from functools import lru_cache
from typing import Dict, Optional, List
_BASE = "https://hpo.jax.org/api/hpo"
_TIMEOUT = 15
_HEADERS = {"User-Agent": "MedGenesis/1.0 (https://huggingface.co/spaces)"}
# ---------------------------------------------------------------------
# Internal helper with retry
# ---------------------------------------------------------------------
async def _get(url: str, *, retries: int = 3) -> Dict:
delay = 2
last: Optional[httpx.Response] = None
for _ in range(retries):
async with httpx.AsyncClient(timeout=_TIMEOUT, headers=_HEADERS) as cli:
last = await cli.get(url)
if last.status_code == 200:
return last.json()
await asyncio.sleep(delay)
delay *= 2
last.raise_for_status() # type: ignore
# ---------------------------------------------------------------------
# Public helpers (cached)
# ---------------------------------------------------------------------
@lru_cache(maxsize=512)
async def get_hpo(term_id: str) -> Dict:
"""Return JSON for `HP:NNNNNNN` (raises if not found)."""
return await _get(f"{_BASE}/term/{term_id}")
@lru_cache(maxsize=256)
async def search_hpo(term: str) -> Optional[Dict]:
"""Return top search hit (`id`, `name`) for free‑text query or None."""
data = await _get(f"{_BASE}/search/{term}")
hits: List[Dict] = data.get("_embedded", {}).get("terms", [])
return hits[0] if hits else None
# ---------------------------------------------------------------------
# CLI test
# ---------------------------------------------------------------------
if __name__ == "__main__":
async def _demo():
hit = await search_hpo("microcephaly")
print("Top search hit:", hit)
if hit:
full = await get_hpo(hit["id"])
print("Definition:", full.get("definition"))
asyncio.run(_demo())
|