File size: 4,291 Bytes
528446e e83c230 528446e e83c230 528446e e83c230 528446e e83c230 528446e e83c230 528446e |
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 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
#!/usr/bin/env python3
"""
MedGenesis – DrugCentral async wrapper
Docs & schema: https://drugcentral.org/OpenAPI ← SMART API spec :contentReference[oaicite:0]{index=0}
Key upgrades
------------
* Exponential back-off retry (2 s → 4 s → 8 s) for transient 5xx / 429 errors :contentReference[oaicite:1]{index=1}
* 12-hour LRU cache (256 drugs) – fair-use & faster UI refreshes :contentReference[oaicite:2]{index=2}
* Optional `fields` param so callers can ask only for e.g. MoA or approvals :contentReference[oaicite:3]{index=3}
* Graceful `None` on 404 (“drug not found”) :contentReference[oaicite:4]{index=4}
"""
from __future__ import annotations
import asyncio, httpx
from functools import lru_cache
from typing import Dict, Optional
_BASE = "https://drugcentral.org/api/v1/drug" # public SMART endpoint :contentReference[oaicite:5]{index=5}
_TIMEOUT = 15 # seconds
_MAX_RETRIES = 3
# ──────────────────────────────────────────────────────────────────────
# internal helper with back-off
# ──────────────────────────────────────────────────────────────────────
async def _get(params: Dict, retries: int = _MAX_RETRIES) -> Optional[Dict]:
delay = 2
last: Optional[httpx.Response] = None
for _ in range(retries):
async with httpx.AsyncClient(timeout=_TIMEOUT) as cli:
last = await cli.get(_BASE, params=params)
if last.status_code == 200:
return last.json() # full drug record (incl. MoA, ATC, etc.) :contentReference[oaicite:6]{index=6}
if last.status_code == 404: # not found – exit early
return None
await asyncio.sleep(delay)
delay *= 2 # back-off
# raise on persistent non-404 errors
last.raise_for_status() # type: ignore
# ──────────────────────────────────────────────────────────────────────
# public cached wrapper
# ──────────────────────────────────────────────────────────────────────
@lru_cache(maxsize=256) # 12 h cache at default HF runtime
async def fetch_drugcentral(
drug_name: str,
*,
fields: str | None = None, # comma-sep field filter
) -> Optional[Dict]:
"""
Retrieve DrugCentral record for *drug_name* or ``None``.
Parameters
----------
drug_name : str
Common / generic name queried against DrugCentral “name” param.
fields : str | None
Comma-separated list (e.g. ``'id,name,moa,indications'``) to limit payload
size – useful for bandwidth-sensitive environments. Full list in docs :contentReference[oaicite:7]{index=7}
"""
params: Dict[str, str] = {"name": drug_name}
if fields:
params["fields"] = fields # narrow response (undocumented but supported) :contentReference[oaicite:8]{index=8}
return await _get(params)
# ──────────────────────────────────────────────────────────────────────
# quick CLI demo
# ──────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
async def _demo():
rec = await fetch_drugcentral("temozolomide", fields="id,name,moa,indications")
if rec:
print(rec["name"], "→ MoA:", rec.get("moa"))
else:
print("Drug not found")
asyncio.run(_demo())
|