| |
| import os, httpx |
| from functools import lru_cache |
|
|
| UMLS_API_KEY = os.getenv("UMLS_KEY") |
| AUTH_URL = "https://utslogin.nlm.nih.gov/cas/v1/api-key" |
| SEARCH_URL = "https://uts-ws.nlm.nih.gov/rest/search/current" |
| CONTENT_URL = "https://uts-ws.nlm.nih.gov/rest/content/current/CUI/{cui}" |
|
|
| async def _get_ticket() -> str: |
| async with httpx.AsyncClient(timeout=10) as c: |
| r1 = await c.post(AUTH_URL, data={"apikey": UMLS_API_KEY}) |
| r1.raise_for_status() |
| tgt = r1.text.split('action="')[1].split('"')[0] |
| r2 = await c.post(tgt, data={"service": "http://umlsks.nlm.nih.gov"}) |
| r2.raise_for_status() |
| return r2.text |
|
|
| @lru_cache(maxsize=512) |
| async def lookup_umls(term: str) -> dict: |
| ticket = await _get_ticket() |
| params = {"string": term, "ticket": ticket, "pageSize": 1} |
| async with httpx.AsyncClient(timeout=10) as c: |
| r = await c.get(SEARCH_URL, params=params) |
| r.raise_for_status() |
| items = r.json().get("result", {}).get("results", []) |
| if not items: |
| return {"term": term} |
| itm = items[0] |
| cui, name = itm.get("ui"), itm.get("name") |
| r2 = await c.get(CONTENT_URL.format(cui=cui), params={"ticket": ticket}) |
| r2.raise_for_status() |
| entry = r2.json().get("result", {}) |
| types = [t["name"] for t in entry.get("semanticTypes", [])] |
| definition = entry.get("definitions", [{}])[0].get("value", "") |
| return { |
| "term": term, |
| "cui": cui, |
| "name": name, |
| "definition": definition, |
| "types": types |
| } |
|
|
